流式传输到 SQL 数据仓库时 Azure Databricks 缺少条目答案

【问题标题】：Azure Databricks Missing entries when streaming to SQL Datawarehouse流式传输到 SQL 数据仓库时 Azure Databricks 缺少条目
【发布时间】：2019-03-31 21:29:40
【问题描述】：

首先我有以下说明，当上传 20.000 个文件时，我在数据库中获得了 20.000 条记录（每个文件仅包含 1 个记录）。

aTracking = sqlContext.read.format('csv').options(header='true', delimiter=';').schema(csvSchema).load("wasbs://" + blobContainer + "@" + blobStorage + ".blob.core.windows.net/rtT*.csv")

aTracking.write \
    .option('user', dwUser) \
    .option('password', dwPass) \
    .jdbc('jdbc:sqlserver://' + dwServer + ':' + dwJdbcPort + ';database=' + dwDatabase, 'stg_tr_energy_xmlin.csv_in', mode = 'append' )

然后，出于速度目的，我认为使用 Polybase 进行流式传输会更好……编码为……但我只有 +- 17.000 个条目。

aTracking = spark.readStream.format('csv').options(header='true', delimiter=';').schema(csvSchema).load("wasbs://" + blobContainer + "@" + blobStorage + ".blob.core.windows.net/rtT*.csv")

aTracking.writeStream \
         .format("com.databricks.spark.sqldw") \
         .option("url", sqlDwUrl) \
         .option("tempDir", "wasbs://uploaddw@" + blobStorage + ".blob.core.windows.net/stream") \
         .option("forwardSparkAzureStorageCredentials", "true") \
         .option("dbTable", "stg_tr_energy_xmlin.csv_in") \
         .option("checkpointLocation", "/checkpoint") \
         .start()

有什么可能导致这种情况的建议吗？

【问题讨论】：

第一步是确定哪些缺失并尝试辨别模式
查明发生了什么。显然系统会提醒已经处理了哪些文件。在第一次运行中，我只加载了 3000 个文件；重新启动流式传输、截断表并上传 20.000 个文件。 +-3000 同名...
任何人都知道它保存在哪里，流文件的“列表”？以后删除它们会很有用。

标签： spark-streaming databricks azure-sqldw

【解决方案1】：

在检查点位置跟踪结构化流式查询的状态。 “假设每个流源都有偏移量（类似于 Kafka 偏移量（...））来跟踪流中的读取位置。引擎使用检查点和预写日志来记录每个触发器中正在处理的数据的偏移量范围”。有关详细信息，请参阅Spark documentation（搜索检查点）。

因此，如果您想重新处理所有文件，请删除以下定义的检查点位置目录（或定义一个新目录）：

.option("checkpointLocation", "/checkpoint").

另外还检查了目标目录中的 _spark_metadata 文件，因此如果要再次写入所有数据，还应该清理目标目录（使用 Azure SQL 数据仓库临时目录）。

【讨论】：