找不到数据源：Python环境中的delta

Posted 2023-04-15

【中文标题】找不到数据源：Python环境中的delta

【英文标题】：Failed to find data source: delta in Python environment

【发布时间】：2021-12-02 15:59:51

【问题描述】：

关注：https://docs.delta.io/latest/quick-start.html#python

我已经安装了 delta-spark 并运行：

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 = spark = configure_spark_with_delta_pip(builder).getOrCreate()

但是当我跑步时：

data = spark.range(0, 5)
data.write.format("delta").save("/tmp/delta-table")

错误状态：无法识别增量

&如果我跑步

DeltaTable.isDeltaTable(spark, "packages/tests/streaming/data")

它指出：TypeError: 'JavaPackage' 对象不可调用

似乎我可以在没有 Maven 的情况下在本地运行这些命令（例如单元测试）或在 pyspark shell 中运行它？看看我是否缺少依赖项会很好吗？

【参考方案1】：

您可以使用pip install delta-spark 安装delta-spark PyPi 包（它也会拉取pyspark），然后参考它。

或者您可以添加一个配置选项来获取 Delta 包。这是.config("spark.jars.packages", "io.delta:delta-core_2.12:<delta-version>")。对于 Spark 3.1 Delta 版本是 1.0.0（有关详细信息，请参阅 releases mapping docs）。

我有一个example of using Delta tables in unit tests（请注意，该 import 语句在函数定义中，因为 Delta 包是动态加载的）：

import pyspark
import pyspark.sql
import pytest
import shutil
from pyspark.sql import SparkSession

delta_dir_name = "/tmp/delta-table"

@pytest.fixture
def delta_setup(spark_session):
    data = spark_session.range(0, 5)
    data.write.format("delta").save(delta_dir_name)
    yield data
    shutil.rmtree(delta_dir_name, ignore_errors=True)

def test_delta(spark_session, delta_setup):
    from delta.tables import DeltaTable
    deltaTable = DeltaTable.forPath(spark_session, delta_dir_name)
    hist = deltaTable.history()
    assert hist.count() == 1

环境是initialized via pytest-spark:

[pytest]
filterwarnings =
  ignore::DeprecationWarning
spark_options =
  spark.sql.extensions: io.delta.sql.DeltaSparkSessionExtension
  spark.sql.catalog.spark_catalog: org.apache.spark.sql.delta.catalog.DeltaCatalog
  spark.jars.packages: io.delta:delta-core_2.12:1.0.0
  spark.sql.catalogImplementation: in-memory

【讨论】：

谢谢 Alex，我在清除我的 pipenv 环境的缓存然后安装 pyspark=3.1.0 后发现它工作了