Databricks Delta实时表-应用来自Delta表的更改

Question

我正在使用Databricks Delta Live表，但是在向上插入一些表时遇到了一些问题。我知道下面的文字很长，但我还是尽可能清晰地描述了我的问题。如果有些部分不清楚，请告诉我。

我有下面的表格和流程:

着陆区（_Z）-

要从着陆区转到原始表，我有以下Pysark代码：

cloudfile = {"cloudFiles.format":"JSON", 
                       "cloudFiles.schemaLocation": sourceschemalocation, 
                       "cloudFiles.inferColumnTypes": True}

@dlt.view('landing_view')
def inc_view():
    df = (spark
             .readStream
             .format('cloudFiles')
             .options(**cloudFilesOptions)
             .load(filpath_to_landing)
     <Some transformations to go from JSON to tabular (explode, ...)>
     return df

dlt.create_target_table('raw_table', 
                        table_properties = {'delta.enableChangeDataFeed': 'true'})
  
dlt.apply_changes(target='raw_table',
                  source='landing_view',
                  keys=['id'],
                  sequence_by='updated_at')

此代码按预期方式工作。我运行它，添加更改。JSON文件到登陆区域，重新运行管道，更新插入正确应用于“raw_table”

(但是，每次在delta文件夹中创建一个包含所有数据的新的parquet文件时，我希望只添加一个包含插入和更新行的parquet文件。关于当前版本的一些信息保存在增量日志中？不确定这是否与我的问题相关。我已经将' raw_table '的table_properties更改为enableChangeDataFeed = true。“intermediate_table”的readStream则具有option(readChangeFeed，“true”)。

然后，我有下面的代码从我的“raw_table”转到我的“intermediate_table”:

@dlt.table(name='V_raw_table', table_properties={delta.enableChangeDataFeed': 'True'})
def raw_table():
     df = (spark.readStream
                .format('delta')
                .option('readChangeFeed', 'true')
                .table('LIVE.raw_table'))
     df = df.withColumn('ExtraCol', <Transformation>)
     return df
 ezeg
dlt.create_target_table('intermediate_table')
dlt.apply_changes(target='intermediate_table',
                  source='V_raw_table',
                  keys=['id'],
                  sequence_by='updated_at')

不幸的是，当我运行这个命令时，我得到了一个错误:‘在版本2的源表中检测到一个数据更新(例如part-00000-7127 BD 29-6820-406 c-a5a 1-e76fc 7126150-c000 . snappy . parquet)。目前不支持此功能。如果您想忽略更新，请将选项“ignoreChanges”设置为“true”。如果您希望反映数据更新，请使用新的检查点目录重新启动此查询。

我签入了“忽略更改”，但不要认为这是我想要的。我希望自动加载器能够检测到增量表中的更改，并将它们传递到流中。

我知道readStream只与append一起工作，但这就是为什么我希望在“raw_table”被更新后，一个新的parquet文件将被添加到delta文件夹中，只有插入和更新。然后，这个添加的拼花文件被自动加载器检测到，并可用于将更改应用到“intermediate_table”中。

我这样做不对吗？还是我忽略了什么？提前感谢！

Answer 1

当您更改设置以包含选项“. ption”（“readChangeFeed”，“true”）“时，您应该从完全刷新开始（开始附近有下拉列表）。这样做将解决错误“检测到的数据更新xxx”，您的代码应该适用于增量更新。

Answer 2

由于 readStream 仅适用于追加，因此源文件中的任何更改都会在下游产生问题。假设“raw_table”上的更新只会插入新的 parquet 文件是不正确的。根据“优化写入”等设置，甚至不带“优化写入”，apply_changes可以添加或删除文件。您可以在“raw_table/_delta_log/xxx.json”中的“num目标文件已添加”和“已删除目标文件”下找到此信息。

基本上，“Databricks建议您使用自动加载程序仅引入不可变文件”。