NRL-1187 deal with timestamps

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 flatten_df, logSchema
+from transformations import dtype_conversion, 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=[flatten_df],
+    transformations=[flatten_df, dtype_conversion],
 )
 
 # Run the job

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

@@ -1,3 +1,4 @@
+from pyspark.sql.functions import to_timestamp
 from pyspark.sql.types import (
     BooleanType,
     StringType,
@@ -19,7 +20,7 @@
                     StructField("level", StringType(), True),
                     StructField("location", StringType(), True),
                     StructField("message", StringType(), True),
-                    StructField("timestamp", TimestampType(), True),
+                    StructField("timestamp", StringType(), True),
                     StructField("service", StringType(), True),
                     StructField("cold_start", BooleanType(), True),
                     StructField("function_name", StringType(), True),
@@ -68,5 +69,8 @@ def flatten_df(df):
     return df.select(*cols, f"{nested_col}.*")
 
 
-def placeholder(df):
+def dtype_conversion(df):
+    df = df.withColumn(
+        "timestamp", to_timestamp(df["timestamp"], "yyyy-MM-dd HH:mm:ss,SSSXXX")
+    )
     return df