在 MLflow 项目中通过 Spark 访问 Databricks 中的 Delta Lake 表

Question

我目前正在使用 spark 从 databricks 笔记本访问 deltalake 表。但是现在我需要从 MLflow 项目访问增量表。 MLflow spark api 只允许记录和加载 SparkML 模型。关于如何实现这一点的任何想法？

目前我正在尝试通过 MLflow 项目中的此代码访问 spark：

spark = pyspark.sql.SparkSession._instantiatedSession
if spark is None:
  # NB: If there is no existing Spark context, create a new local one.
  # NB: We're disabling caching on the new context since we do not need it and we want to
  # avoid overwriting cache of underlying Spark cluster when executed on a Spark Worker
  # (e.g. as part of spark_udf).
  spark = ( pyspark.sql.SparkSession.builder \
   .config("spark.python.worker.reuse", True)
   .config("spark.databricks.io.cache.enabled", False)
   # In Spark 3.1 and above, we need to set this conf explicitly to enable creating
   # a SparkSession on the workers
   .config("spark.executor.allowSparkContext", "true")
   .master("local[*]")
   .appName("MLflow Project")
   .getOrCreate()
  )

但我收到此错误：

py4j.protocol.Py4JJavaError: An error occurred while calling None.org.apache.spark.api.java.JavaSparkContext.

Answer 1

在 Databricks 笔记本中运行 MLflow 项目（例如，针对连接到该集群的笔记本中的现有交互式集群）目前没有得到很好的支持，原因有多种（例如，缺少对创建的子进程的身份验证传播）运行项目）。

Answer 2

应该与不在 Notebooks 中运行的普通 Spark 项目以相同的方式完成：

• 为spark-submit或pyspark添加依赖，需要安装delta-spark包才能使用代码补全等（--conf可以从代码本身设置，看下一步）：

pyspark --packages io.delta:delta-core_2.12:1.1.0 \
  --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
  --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"

• 创建一个SparkSession 对象并使用它：

import pyspark
from delta import *

builder = pyspark.sql.SparkSession.builder.appName("MyApp") \
  .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
  .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")

spark = configure_spark_with_delta_pip(builder).getOrCreate()

这些步骤包含在 Delta 文档中的 Quickstart guide 中。

如果数据驻留在 Azure Data Lake Storage、AWS S3 或 GCP 上，您可能需要添加额外的包和配置，但这也包含在文档中。