我刚开始使用delta lake,所以我的心理模型可能不适用-我问这个问题是为了验证/反驳它。
我对delta lake的理解是,它只存储对数据的增量更改(“delta”)。有点像git—每次提交时,并不是存储整个代码库的快照—提交只包含所做的更改。类似地,我可以想象,如果我创建一个delta表,然后尝试用表中已经包含的所有内容“更新”该表(即“空提交”),那么我就不会期望看到由于该更新而创建的任何新数据。
然而,这并不是我所观察到的:这样的更新似乎复制了现有的表。发生什么事?这在我看来不是很“增量”。
(为了可读性,我将替换文件名中的实际uuid值)
# create the data
dataGen = sc._jvm.org.apache.hudi.QuickstartUtils.DataGenerator()
inserts = sc._jvm.org.apache.hudi.QuickstartUtils.convertToStringList(dataGen.generateInserts(200))
df = spark.read.json(spark.sparkContext.parallelize(inserts, 2))
delta_base_path = "s3://my-delta-table/test"
(df.write.format("delta")
.mode("overwrite")
.save(delta_base_path))
!aws s3 ls --recursive s3://my-delta-table/test
2020-10-19 14:33:21 1622 test/_delta_log/00000000000000000000.json
2020-10-19 14:33:18 0 test/_delta_log_$folder$
2020-10-19 14:33:19 10790 test/part-00000-UUID1-c000.snappy.parquet
2020-10-19 14:33:19 10795 test/part-00001-UUID2-c000.snappy.parquet
然后我使用完全相同的数据集覆盖:
(df.write.format("delta")
.mode("overwrite")
.save(delta_base_path))
!aws s3 ls --recursive s3://my-delta-table/test
2020-10-19 14:53:02 1620 test/_delta_log/00000000000000000000.json
2020-10-19 14:53:08 872 test/_delta_log/00000000000000000001.json
2020-10-19 14:53:01 0 test/_delta_log_$folder$
2020-10-19 14:53:07 6906 test/part-00000-UUID1-c000.snappy.parquet
2020-10-19 14:53:01 6906 test/part-00000-UUID3-c000.snappy.parquet
2020-10-19 14:53:07 6956 test/part-00001-UUID2-c000.snappy.parquet
2020-10-19 14:53:01 6956 test/part-00001-UUID4-c000.snappy.parquet
现在有两个 part-0
以及 part-1
s、 大小完全一样。为什么增量重复数据不删除现有数据?
1条答案
按热度按时间liwlm1x91#
这并不是完全正确的理解-delta不会自动检查现有数据的重复项-如果您只想存储新的/更新的数据,那么您需要使用
merge
操作,该操作将检查现有数据,然后您可以决定如何处理现有数据—用新数据覆盖,或者忽略该操作。你可以在delta的网站上找到更多的信息,或者在learning spark的第9章,2ed书籍中找到(可以从databricks免费获得)