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
1 ACCEPTED SOLUTION
Hi @Retired_mod , when running the code on the dlt pipeline, we don't get to choose what cluster the dlt pipeline will use. The dlt pipeline just stands up a random temporary cluster that uses for the dlt run. Apologies if technically I am off but from a high level that's the way I see the dlt pipeline working.
The only place I see where the dlt pipeline can be configured with additional libraries is within the dlt pipeline code itself by installing with the syntax as# MAGIC %pip install mlflow , etc... I've tried to resolve the error also by running the code with the added # MAGIC %pip install importlib_metadata (as shown in my code on my post) but no luck.
as the documentation states, installing of the mlflow library has to be done within the notebook of the dlt pipeline run.
