Add throughput logging

ex_cubic_ingestion/lib/ex_cubic_ingestion/process_incoming.ex

@@ -14,6 +14,7 @@ defmodule ExCubicIngestion.ProcessIncoming do
   alias ExCubicIngestion.Schema.CubicLoad
   alias ExCubicIngestion.Schema.CubicTable
   alias ExCubicIngestion.Validators
+  require Logger
 
   @wait_interval_ms 5_000
 
@@ -66,17 +67,31 @@ defmodule ExCubicIngestion.ProcessIncoming do
       |> CubicTable.filter_to_existing_prefixes()
 
     for {table_prefix, table} <- table_prefixes do
-      incoming_bucket
-      |> S3Scan.list_objects_v2(
-        prefix: "#{incoming_prefix}#{table_prefix}",
-        lib_ex_aws: state.lib_ex_aws
-      )
-      # filter s3 objects to only data objects with a size specified
-      |> Enum.filter(&Validators.valid_s3_object?(&1))
-      |> Enum.map(fn object ->
-        %{object | key: String.replace_prefix(object[:key], incoming_prefix, "")}
-      end)
-      |> CubicLoad.insert_new_from_objects_with_table(table)
+      result =
+        incoming_bucket
+        |> S3Scan.list_objects_v2(
+          prefix: "#{incoming_prefix}#{table_prefix}",
+          lib_ex_aws: state.lib_ex_aws
+        )
+        # filter s3 objects to only data objects with a size specified
+        |> Enum.filter(&Validators.valid_s3_object?(&1))
+        |> Enum.map(fn object ->
+          %{object | key: String.replace_prefix(object[:key], incoming_prefix, "")}
+        end)
+        |> CubicLoad.insert_new_from_objects_with_table(table)
+
+      case result do
+        {:ok, {_snapshot, [_ | _] = new_loads}} ->
+          Logger.info(
+            "parent=data_platform, process=run, table=#{table.name}, " <>
+              "s3_prefix=#{table.s3_prefix}, file_count=#{length(new_loads)}, "
+          )
+
+        _ ->
+          :ok
+      end
+
+      result
     end
 
     :ok

py_cubic_ingestion/lib/py_cubic_ingestion/ingest_incoming.py

@@ -9,6 +9,7 @@
 from py_cubic_ingestion import job_helpers
 from pyspark.context import SparkContext
 import boto3
+import logging
 import sys
 
 
@@ -59,4 +60,13 @@ def run() -> None:
         # write out to springboard bucket using the same prefix as incoming
         job_helpers.write_parquet(updated_table_df, load.get("partition_columns", []), load["destination_path"])
 
+        logging.info(
+            "parent=data_platform, process=ingest_incoming, source_table=%s, destination_table=%s, "
+            "destination_path=%s, source_s3_key=%s, row_count=%d, col_count=%d, "
+            "partition_count=%d",
+            load["source_table_name"],
+            load["destination_table_name"],
+            load["destination_path"],
+            load["source_s3_key"]
+        )
     job.commit()