I am running this notebook via the dlt pipeline in preview mode.
everything works up until the predictions table that should be created with a registered model inferencing the gold table.
This is the error: com databricks spark safespark UDFException: INVALID_ARGUMENT: No module named 'importlib_metadata'
# Databricks notebook source
# MAGIC %pip install mlflow
# MAGIC %pip install importlib_metadata
# COMMAND ----------
import mlflow
import importlib_metadata
model_uri = f"models:/soybeans_volatility/1"
# create spark user-defined function for model prediction.
predict = mlflow.pyfunc.spark_udf(spark, model_uri, result_type="double", env_manager='virtualenv')
# COMMAND ----------
import dlt
from pyspark.sql.functions import avg, max, min, col, lag, count, when, struct, from_unixtime, unix_timestamp
from pyspark.sql.window import Window
path_to_uc_external_location = "s3://gfy-databricks-storage/data/barchart/soybeans/"
@dlt.table(name="soybeans_bronze", table_properties={"quality": "bronze"})
def table_name():
return (
spark.readStream
.format("cloudFiles")
.option("cloudFiles.format", "json")
.load(f"{path_to_uc_external_location}")
)
@dlt.table(name="soybeans_silver", table_properties={"quality": "silver"})
def create_silver_table():
df = dlt.read("soybeans_bronze")
cleaned_df = df.drop("_rescued_data").filter(col("close").isNotNull())
formatted_df = cleaned_df.withColumn(
"tradeTimestamp",
from_unixtime(
unix_timestamp(col("timestamp"), "yyyy-MM-dd'T'HH:mm:ssXXX"),
"yyyy-MM-dd HH:mm:ss.SSS"
)
)
return formatted_df
@dlt.table(name="soybeans_gold", table_properties={"quality": "gold"})
def create_gold_table():
df_silver = dlt.read("soybeans_silver")
# Compute a 7-day rolling average of the close price
windowSpec = Window.partitionBy("symbol").orderBy("tradeTimestamp").rowsBetween(-6, 0)
avg_price = avg(col("close").cast("double")).over(windowSpec)
# Compute daily volatility
daily_volatility = (max(col("high").cast("double")).over(windowSpec) -
min(col("low").cast("double")).over(windowSpec))
# Extract previous day's volume
lag_window = Window.partitionBy("symbol").orderBy("tradeTimestamp").rowsBetween(-1, -1)
prev_day_volume = lag(col("volume"), 1, 0).over(lag_window)
df_gold = df_silver.withColumn("7_day_avg_close", avg_price) \
.withColumn("daily_volatility", daily_volatility) \
.withColumn("prev_day_volume", prev_day_volume)
return df_gold
@dlt.table(
comment="DLT for predictions scored by soybeans_volatility model based on models.soybeans.soybeans_gold Delta table.",
name="soybeans_gold_preds",
table_properties={
"quality": "gold"
}
)
def soybeans_volatility_predictions():
input_dlt_table_name = "soybeans_gold"
input_delta_live_table = dlt.read(input_dlt_table_name)
input_dlt_table_columns = input_delta_live_table.columns
predictions_df = input_delta_live_table.withColumn('prediction', predict(struct(*input_dlt_table_columns)))
return predictions_df
I've tried everything, I've removed the virtualenv and ran the pipeline in current mode (non preview) but no luck.
This is what I'm using as a guide: https://docs.databricks.com/en/delta-live-tables/transform.html
