读取文件夹中的增量表特定文件

Question

我正在尝试从包含多个增量文件的文件夹中读取特定文件，请参阅随附的屏幕截图

原因我希望根据架构版本读取增量文件。上述文件夹包含具有不同模式结构的文件。

用于写入文件的代码片段：

df.write.format("delta").mode("overwrite").option("overwriteSchema", "true").save("/home/games/Documents/test_delta/")

读取增量文件的代码

import pyspark[![enter image description here][1]][1]

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()

path_to_data = '/home/games/Documents/test_delta/_delta_log/00000000000000000001.json'
df = spark.read.format("delta").load(path_to_data)
df.show()

错误：

org.apache.spark.sql.delta.DeltaAnalysisException：

/home/games/Documents/test_delta/_delta_log/

不是 Delta 表。

Answer 1

您应该使用：

df = spark.read.format("delta").option("versionAsOf", 0).load(path_to_data)

您可以指定其他版本而不是 0，具体取决于覆盖数据的次数。您还可以使用时间戳。请参阅delta 快速入门了解更多信息。

此外，delta_log 文件夹实际上包含 json 格式的 delta transaction log，而不是实际数据。数据存在于父文件夹中（在您的情况下为 test_delta ）。以 part-0000 开头的文件是包含实际数据的文件。这些是 .parquet 文件。没有带有 .delta 扩展名的文件。

Answer 2

如果您想查看实际数据，您应该搜索“test_delta”。这意味着，

path_to_data = '/home/games/Documents/test_delta'
df = spark.read.format("delta").load(path_to_data)
df.show()

当您更新或删除某些内容时，

delta_log 部分会自行更新，并将新的 json 文件添加为 0000...1.json 或 0000...2.json。感谢 json 文件，Delta Lake 获得了包含 part-0000-...parquet.

的数据