标签: databricks-autoloader

Databricks Delta Live 表 - 应用增量表中的更改

我正在使用 Databricks Delta Live Tables,但在向上游插入某些表时遇到一些问题。我知道下面的文字很长,但我试图尽可能清楚地描述我的问题。如果某些部分不清楚,请告诉我。

我有以下表格和流程:

Landing_zone -> 这是一个添加 JSON 文件的文件夹,其中包含插入或更新的记录的数据。Raw_table -> 这是 JSON 文件中的数据,但采用表格格式。该表采用增量格式。除了将 JSON 结构转换为表格结构(我进行了爆炸,然后从 JSON 键创建列)之外,未进行任何转换。Intermediate_table -> 这是 raw_table,但有一些额外的列(取决于其他列值)。

为了从我的着陆区转到原始表,我有以下 Pyspark 代码:

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')
Run Code Online (Sandbox Code Playgroud)

这段代码按预期工作。我运行它,将一个changes.JSON文件添加到登陆区域,重新运行管道,并将更新插入正确地应用于“raw_table”

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

然后我有以下代码从“raw_table”转到“intermediate_table”: …

databricks delta-lake databricks-autoloader delta-live-tables

7
推荐指数
1
解决办法
7742
查看次数

我们可以从 Databricks Autoloader 中排除或仅包含特定文件扩展名吗?

现在,databricks 自动加载器需要一个从中加载所有文件的目录路径。但是,如果某些其他类型的日志文件也开始进入该目录 - 有没有办法要求 Autoloader 在准备数据帧时排除这些文件?

df = spark.readStream.format("cloudFiles") \
  .option(<cloudFiles-option>, <option-value>) \
  .schema(<schema>) \
  .load(<input-path>)
Run Code Online (Sandbox Code Playgroud)

databricks databricks-autoloader

2
推荐指数
1
解决办法
2230
查看次数