refactor ingestion script to use iceberg tables

Author: vitor-pina

Docker/Dockerfile

@@ -1,5 +1,5 @@
 FROM nauedu/nau-analytics-base-spark:latest
-
+RUN pip install --no-cache-dir git+https://github.com/fccn/nau-analytics-utils.git@main#subdirectory=common_libs/utils
 # Copy your application code
 COPY src/ /opt/spark/work-dir/src/
 

src/bronze/get_full_tables.py src/bronze/python/get_full_tables.pysrc/bronze/get_full_tables.py renamed to src/bronze/python/get_full_tables.py

@@ -1,13 +1,11 @@
-from pyspark.sql import SparkSession #type:ignore
 from pyspark.sql import DataFrame #type:ignore
-from pyspark.sql.types import StructType, StructField, StringType, IntegerType, FloatType #type:ignore
+from nau_analytics_data_product_utils_lib import Config,get_required_env,get_iceberg_spark_session #type: ignore
 import pyspark.sql.functions as F #type:ignore
 import argparse
 import os
 import logging
 from typing import List, Union, Optional,Tuple
-from utils.utils import Utils
-from utils.config import Config
+
 
 logging.basicConfig(
     level=logging.INFO,
@@ -17,12 +15,11 @@
     ]
 )
 
-utils_obj = Utils("dev")
+
 
 
 def get_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser()
-    parser.add_argument("--savepath", type = str,required= True, help = "The S3 bucket intended for the data to be stored")
     parser.add_argument("--undesired_column", type = str,required= True, help = " the undesired column for a table")
     args = parser.parse_args()
     return args
@@ -32,32 +29,29 @@ def add_ingestion_metadata_column(df: DataFrame,table: str) -> DataFrame:
     tmp_df = df.withColumn("ingestion_date", F.current_timestamp()).withColumn("source_name", F.lit(table))
     return tmp_df
 
-def add_date_partition_columns(df: DataFrame,column_name:str) -> DataFrame:
-    df = df.withColumn("year", F.year(F.col(column_name)))\
-        .withColumn("month", F.month(F.col(column_name)))\
-        .withColumn("day",F.day(column_name))
-    return df
 
 def main() -> None:
 
-    MYSQL_DATABASE = utils_obj.get_required_env("MYSQL_DATABASE")
-    MYSQL_HOST = utils_obj.get_required_env("MYSQL_HOST")
-    MYSQL_PORT = utils_obj.get_required_env("MYSQL_PORT")
-    MYSQL_USER = utils_obj.get_required_env("MYSQL_USER")
-    MYSQL_SECRET = utils_obj.get_required_env("MYSQL_SECRET")
+    MYSQL_DATABASE = get_required_env("MYSQL_DATABASE")
+    MYSQL_HOST = get_required_env("MYSQL_HOST")
+    MYSQL_PORT = get_required_env("MYSQL_PORT")
+    MYSQL_USER = get_required_env("MYSQL_USER")
+    MYSQL_SECRET = get_required_env("MYSQL_SECRET")
     jdbc_url = f"jdbc:mysql://{MYSQL_HOST}:{MYSQL_PORT}/{MYSQL_DATABASE}"
 
-    S3_ACCESS_KEY = utils_obj.get_required_env("S3_ACCESS_KEY")
-    S3_SECRET_KEY = utils_obj.get_required_env("S3_SECRET_KEY")
-    S3_ENDPOINT = utils_obj.get_required_env("S3_ENDPOINT")
+    S3_ACCESS_KEY = get_required_env("S3_ACCESS_KEY")
+    S3_SECRET_KEY = get_required_env("S3_SECRET_KEY")
+    S3_ENDPOINT = get_required_env("S3_ENDPOINT")
     args = get_args()
-    S3_SAVEPATH = args.savepath
     undesired_column = args.undesired_column
 
-    ICEBERG_CATALOG_URI = utils_obj.get_required_env("ICEBERG_CATALOG_URI")
-    ICEBERG_CATALOG_USER = utils_obj.get_required_env("ICEBERG_CATALOG_USER")
-    ICEBERG_CATALOG_PASSWORD = utils_obj.get_required_env("ICEBERG_CATALOG_PASSWORD")
-    ICEBERG_CATALOG_WAREHOUSE = utils_obj.get_required_env("ICEBERG_CATALOG_WAREHOUSE")
+    ICEBERG_CATALOG_HOST = get_required_env("ICEBERG_CATALOG_HOST")
+    ICEBERG_CATALOG_PORT = get_required_env("ICEBERG_CATALOG_PORT")
+    ICEBERG_CATALOG_NAME = get_required_env("ICEBERG_CATALOG_NAME")
+    ICEBERG_CATALOG_USER = get_required_env("ICEBERG_CATALOG_USER")
+    ICEBERG_CATALOG_PASSWORD = get_required_env("ICEBERG_CATALOG_PASSWORD")
+    ICEBERG_CATALOG_WAREHOUSE = get_required_env("ICEBERG_CATALOG_WAREHOUSE")
+    ICEBERG_CATALOG_URI = f"jdbc:mysql://{ICEBERG_CATALOG_HOST}:{ICEBERG_CATALOG_PORT}/{ICEBERG_CATALOG_NAME}"
 
     TABLES = [
     "course_overviews_courseoverview", 
@@ -82,7 +76,7 @@ def main() -> None:
         iceberg_catalog_warehouse=ICEBERG_CATALOG_WAREHOUSE
 
     )
-    spark = utils_obj.get_iceberg_spark_session(cfg=icerberg_cfg)
+    spark = get_iceberg_spark_session(cfg=icerberg_cfg)
     for table in TABLES:
 
         logging.info(f"getting table {table}")
@@ -95,19 +89,22 @@ def main() -> None:
                 .option("driver", "com.mysql.cj.jdbc.Driver") \
                 .option("dbtable", table) \
                 .load()
+            
             if table == "auth_user":
                 df = df.drop(undesired_column)
 
+
+            
             df = add_ingestion_metadata_column(df=df,table=table)
-            df = add_date_partition_columns(df,"ingestion_date")
+
             if table == "auth_user" and undesired_column and undesired_column in df.columns:
                 raise Exception("THE undesired column stills in the dataframe")
 
-            output_path = f"{S3_SAVEPATH}/{table}"
 
-            df.write.format("iceberg").mode("append").partitionBy("year", "month","day").save(output_path)
+            saveTable = f"bronze_local.entidades.{table}"
+            df.write.format("iceberg").mode("append").saveAsTable(saveTable)
 
-            logging.info(f"Data saved as Delta table to {output_path}")
+            logging.info(f"Data saved as Delta table to {saveTable}")
 
         except Exception as e:
             logging.error(f"Pipeline failed: {e}")