时间戳

首页 > 解决方案 > Koalas / pyspark 找不到数据源：delta

apache-spark - Koalas / pyspark 找不到数据源：delta

问题描述

当我尝试在本地使用 koalas.DataFrame.to_delta() 将 Koalas DataFrame 直接写入 delta 表时，我得到以下 Pyspark 异常：
java.lang.ClassNotFoundException: Failed to find data source: delta
编辑：忽略下面，直接调用 Pyspark 也会出现问题。

如果我将 Koalas DataFrame 转换为 Spark DataFrame 然后写入 delta，我似乎没有问题。是否有 Koalas 不知道但 Pyspark 知道的底层库？似乎很奇怪，因为我认为在后台使用了相同的 Pyspark 模块......我应该注意 Koalas to_delta() 方法似乎确实适用于 Databricks，这表明我的本地设置缺少与 Delta 相关的库.

失败的考拉代码：

kdf = ks.DataFrame({'eid': [1, 2, 3],
                        'contigName': ['chr1', 'chr2', 'chr3'],
                        'phen1': [0.123, 0.456, 0.789],
                        'phen2': [0.987, 0.654, 0.321]})
kdf.to_delta(path='tmp/test.delta', mode='overwrite')

编辑：不工作考拉火花到三角洲代码：

kdf = ks.DataFrame({'eid': [1, 2, 3],
                        'contigName': ['chr1', 'chr2', 'chr3'],
                        'phen1': [0.123, 0.456, 0.789],
                        'phen2': [0.987, 0.654, 0.321]})
kdf.to_spark().write.format('delta').mode('overwrite')

另外，Koalas 和 Spark 如何保存到 delta 表之间有什么区别需要注意吗？我有一个相当大的增量表，到目前为止，它一直是使用 Koalas（在 Databricks 上）写入的，但我可能会切换到 spark.write 以使本地测试更容易。在我这样做之前，我想确保两种方法的结果是相同的（我会做一些测试来确认这一点，只是好奇是否有人对切换现有增量表的写入策略有任何其他说明）。

编辑：好的，我猜 Pyspark 实际上也没有保存 delta 表，我忘了将 .save() 添加到 Pyspark .write 调用中，就像一个哑巴一样。所以我想我现在真正的问题是如何在本地运行 Pyspark 时包含 Delta 库/jar，特别是当我在 Pycharm 中运行单元测试时？

标签： apache-sparkpysparkdatabricksdelta-lakespark-koalas

解决方案

您只需要遵循文档。对于交互式 pyspark，它可能是：

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

或使用代码（首先安装包pip install 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()

请注意，不同版本的 Delta 对 Spark 版本有不同的要求 - 检查您的 Spark 版本的兼容性表。

推荐阅读