如何添加可追溯性列自动加载器 - adf集成？

Question

我正在使用 Azure 数据工厂将源数据复制到登陆区域 （adls gen2），然后使用自动加载程序加载到青铜色增量表中。一切都很完美，除了我无法将pipeline_name，runid和trigger_time派生为镶木地板文件中的派生列以及输入源表。

架构此处是使用实际源 sql Server 表架构构建的结构类型，它不涵盖 ADF 中的其他派生列。

sourceFilePath = 'wasbs://[e-mail]/' \
   + domain_name + '/' + database_name + '/' \
   + schema_table_name.replace(database_name+'.','') + '/'
df = (spark
     .readStream
     .format("cloudFiles")
     .options(**cloudFile)
     .schema(schema)
     .option("rescueDataColumn", "_rescued_data")
     .load(sourceFilePath)
     )
# Traceability columns
# from pyspark.sql.functions import *
df = (
  df.withColumn("audit_fileName", input_file_name()) 
    .withColumn("audit_createdTimestamp", current_timestamp())
)

这是写流DF

streamQuery = (df
           .writeStream
           .format("delta")
           .outputMode("append")
           .trigger(once=True)
           .queryName(queryName)
           .option("checkpointLocation",checkpointLocation)
           .option("mergeSchema", "true")
           .start(tablePath)
          )

使用 mergeSchema True - 我期望流在写入增量格式时从数据工厂检测到 3 个额外的列。这是镶木地板的限制吗？我是否以 csv / json 格式读取数据？或者我必须添加派生列架构定义。

Answer 1

我的经验是，当您在readStream中添加列时，模式演化已经完成。我使用的做法是用这些列为表添加种子，并让模式演化处理传入数据。

CREATE Table IF NOT EXISTS db.tbl
(
  file_path               String,
  created_timestamp       Timestamp,
  last_modified_timestamp Timestamp,
  created_userid          String,
  _rescued_data           String
)
USING DELTA...

Answer 2

您可以在readStream命令中添加审核字段：

from pyspark.sql import functions as F

sourceFilePath = 'wasbs://[e-mail]/' \
   + domain_name + '/' + database_name + '/' \
   + schema_table_name.replace(database_name+'.','') + '/'
df = (spark
     .readStream
     .format("cloudFiles")
     .options(**cloudFile)
     .schema(schema)
     .option("rescueDataColumn", "_rescued_data")
     .load(sourceFilePath)
     .withColumn("audit_fileName", input_file_name()) 
     .withColumn("audit_createdTimestamp", current_timestamp())
)
# Just displaying the dataframe with the audit columns:
df.display()