feat(databricks): make databricks save to external tables in azure storage account

databricks/Havvarsel - Ingest and store Depth Index Table.py

@@ -17,5 +17,16 @@
     "depthItem.depthValue"
 )
 
-depth_data.write.format("delta").saveAsTable("havvarsel_depth_index_to_meter_mapping")
+# COMMAND ----------
+
+from helpers.adls_utils import save_df_as_delta
+save_df_as_delta(depth_data, "depth_index_to_meter_mapping")
+
+
+# COMMAND ----------
+
+from helpers.adls_utils import read_df_as_delta
+
+df = read_df_as_delta("depth_index_to_meter_mapping")
+display(df)
 

databricks/Havvarsel - Ingest as Bronze.py

@@ -14,7 +14,6 @@
 url = f"https://api.havvarsel.no/apis/duapi/havvarsel/v2/temperatureprojection/{lat}/{lon}?depth={depth_index}"
 headers = {"accept": "application/json"}
 response = requests.get(url, headers=headers)
-
 data = response.json()
 df_raw = spark.read.json(sc.parallelize([data]))
 
@@ -23,6 +22,7 @@
 
 # COMMAND ----------
 
+from datetime import datetime 
 depth_data = spark.table("havvarsel_depth_index_to_meter_mapping")
 depth_m = depth_data.filter(depth_data.depthIndex == depth_index).collect()[0].depthValue
 fetch_date = datetime.now().strftime("%Y-%m-%d")
@@ -31,12 +31,7 @@
 
 # COMMAND ----------
 
-from datetime import datetime
-
-bronze_fetch_date_path = f"/mnt/data/bronze/hav_temperature_projection_{fetch_date}"
-bronze_latest_data_path = "/mnt/data/bronze/hav_temperature_projection_latest"
-
-df_bronze.write.format("delta").mode("overwrite").save(bronze_fetch_date_path)
-df_bronze.write.format("delta").mode("overwrite").save(bronze_latest_data_path)
-
+from helpers.adls_utils import save_df_as_delta
+save_df_as_delta(df_bronze, f"/bronze/hav_temperature_projection_{fetch_date}")
+save_df_as_delta(df_bronze, "/bronze/hav_temperature_projection_latest")
 

databricks/Havvarsel - Transform to Silver.py

@@ -1,12 +1,12 @@
 # Databricks notebook source
-df_bronze_0 = spark.read.format("delta").load("/mnt/data/bronze/hav_temperature_projection_latest")
+from helpers.adls_utils import read_df_as_delta
+df_bronze_0 = read_df_as_delta("/bronze/hav_temperature_projection_latest")
 display(df_bronze_0)
 
 # COMMAND ----------
 
 from pyspark.sql.functions import explode, from_unixtime, col
 
-
 # Extract lat and lon, and explode variables
 df_bronze_1 = df_bronze_0.select(
     col("closestGridPointWithData.lat").alias("lat"),
@@ -61,11 +61,11 @@
 # COMMAND ----------
 
 
-silver_latest_path = "/mnt/data/silver/hav_temperature_projection_latest"
-df_silver.write.format("delta").mode("overwrite").save(silver_latest_path)
+from helpers.adls_utils import save_df_as_delta
+save_df_as_delta(df_silver, "/silver/hav_temperature_projection_latest")
 
 # COMMAND ----------
 
-df_check_silver = spark.read.format("delta").load(silver_latest_path)
-
+from helpers.adls_utils import read_df_as_delta
+df_check_silver = read_df_as_delta("/silver/hav_temperature_projection_latest")
 display(df_check_silver)

databricks/helpers/adls_utils.py

@@ -0,0 +1,24 @@
+from databricks.sdk.runtime import *
+import pyspark.dbutils
+from pyspark.sql import SparkSession
+
+STORAGE_ACCOUNT = "devaquaplatformst01"
+spark = SparkSession.builder.getOrCreate()
+
+def connect_to_adls(storage_account = STORAGE_ACCOUNT): 
+    spark.conf.set(
+        f"fs.azure.account.key.{storage_account}.dfs.core.windows.net",
+        dbutils.secrets.get(scope="terraform-created-scope", key="storage-account-key"))
+
+def get_adls_file_path(container = "datalake", storage_account = STORAGE_ACCOUNT): 
+    return (f"abfss://{container}@{storage_account}.dfs.core.windows.net/havvarsel/")
+
+
+def save_df_as_delta(df, table_name, mode="overwrite", file_path=get_adls_file_path()): 
+    connect_to_adls()
+    df.write.format("delta").mode(mode).save(f"{file_path}/{table_name}")
+
+def read_df_as_delta(file_name, file_path=get_adls_file_path()): 
+    connect_to_adls()
+    df = spark.read.format("delta").load(f"{file_path}/{file_name}")
+    return df