pyspark 缓慢变化的维度|SCD类型1从先前日期数据中删除行

mbyulnm0  于 5个月前  发布在  Spark
关注(0)|答案(1)|浏览(54)

我已经在Azure数据块中使用merge into语句实现了SCD类型1。
当我第一次加载时,它加载的数据预计有一些只是说5000行,摄取日期为2023年11月28日。
现在,当我今天进行加载并仅从源中提取2023年11月29日的数据时,我的SCD类型1正在加载正确的记录/行计数1300,根据源数据计数是正确的,但现在对于2023年11月28日,记录计数更改为4500。
这是出乎意料的,因为它不应该做任何改变我的旧日期记录。
有没有人可以帮助我理解这个问题?
我使用简单的合并成语句,我更新的情况下,我的关键列之间找到匹配的行和插入记录的情况下,我的关键列之间没有找到匹配。

pxy2qtax

pxy2qtax1#

正如@NotFound所提到的,当您更新表并按时间进行过滤时,您得到的计数小于原始计数,因为某些行已被更新。但是,计数可能比预期的还要小。
在下面的例子中,我有30行,正在更新6行。当过滤出获取此时间戳之前的行时,预期计数为24,但它显示为21。

来源

x1c 0d1x的数据
执行合并后:



预期计数为24,但显示为21。这是因为增量合并重写了某些行而没有修改它们。
在下图中,您可以看到numTargetRowsCopied为3,numOutputRows为9。三条记录被重写而未被修改,这导致计数不同。



您可以参考this解决方案了解更多信息。
我的建议是使用Delta版本控制。您可以使用versiontimestamp获取以前记录的数据,然后您将获得正确的结果。

%sql
select count(*) from target version as of 0 where _metadata.file_modification_time<"2023-12-02T03:57:00"

字符串


相关问题