NRL-1411 Handle empty dataframes

Author: jackleary

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

@@ -104,9 +104,12 @@ def load(self, data):
         self.logger.info(f"Loading data into {self.target_path} as Parquet")
         for name, dataframe in data.items():
             name = name.replace("--", "_")
-            dataframe.coalesce(1).write.mode("append").partitionBy(
-                *self.partition_cols
-            ).parquet(f"{self.target_path}{name}")
+            try:
+                dataframe.coalesce(1).write.mode("append").partitionBy(
+                    *self.partition_cols
+                ).parquet(f"{self.target_path}{name}")
+            except:
+                self.logger.info(f"{name} dataframe has no rows. Skipping.")
 
     def trigger_crawler(self):
         self.glue.start_crawler(Name=f"{self.name_prefix}-log-crawler")

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

@@ -39,19 +39,25 @@ def rename_cols(df):
 
 
 def dtype_conversion(df):
-    df = (
-        df.withColumn(
-            "event_timestamp_cleaned", regexp_replace(col("event_timestamp"), ",", ".")
+    try:
+        df = (
+            df.withColumn(
+                "event_timestamp_cleaned",
+                regexp_replace(col("event_timestamp"), ",", "."),
+            )
+            .withColumn(
+                "event_timestamp",
+                to_timestamp(
+                    col("event_timestamp_cleaned"), "yyyy-MM-dd HH:mm:ss.SSSZ"
+                ),
+            )
+            .withColumn("time", from_unixtime(col("time")).cast("timestamp"))
+            .withColumn("date", to_date(col("time")))
         )
-        .withColumn(
-            "event_timestamp",
-            to_timestamp(col("event_timestamp_cleaned"), "yyyy-MM-dd HH:mm:ss.SSSZ"),
-        )
-        .withColumn("time", from_unixtime(col("time")).cast("timestamp"))
-        .withColumn("date", to_date(col("time")))
-    )
 
-    df = df.drop("event_timestamp_cleaned")
+        df = df.drop("event_timestamp_cleaned")
+    except:
+        ...
 
     select_exprs = []
     for column_name in df.columns: