NRL-1187 format data

Author: jackleary

terraform/account-wide-infrastructure/modules/glue/src/main.py

@@ -3,7 +3,7 @@
 from awsglue.utils import getResolvedOptions
 from pipeline import LogPipeline
 from pyspark.context import SparkContext
-from transformations import logSchema, placeholder
+from transformations import flatten_df, logSchema
 
 # Get arguments from AWS Glue job
 args = getResolvedOptions(sys.argv, ["job_name", "source_path", "target_path"])
@@ -20,7 +20,7 @@
     target_path=args["target_path"],
     schema=logSchema,
     partition_cols=partition_cols,
-    transformations=[placeholder],
+    transformations=[flatten_df],
 )
 
 # Run the job

terraform/account-wide-infrastructure/modules/glue/src/pipeline.py

@@ -50,6 +50,6 @@ def transform(self, dataframe):
     def load(self, dataframe):
         """Load transformed data into Parquet format"""
         self.logger.info(f"Loading data into {self.target_path} as Parquet")
-        dataframe.write.mode("overwrite").partitionBy(*self.partition_cols).parquet(
+        dataframe.write.mode("append").partitionBy(*self.partition_cols).parquet(
             self.target_path
         )

terraform/account-wide-infrastructure/modules/glue/src/transformations.py

@@ -58,5 +58,15 @@
 )
 
 
+def flatten_df(df):
+    cols = []
+    for c in df.dtypes:
+        if "struct" in c[1]:
+            nested_col = c[0]
+        else:
+            cols.append(c[0])
+    return df.select(*cols, f"{nested_col}.*")
+
+
 def placeholder(df):
     return df