databricks autoloader 不立即更新表答案

【问题标题】：databricks autoloader not updating table immediatelydatabricks autoloader 不立即更新表
【发布时间】：2023-01-03 12:36:14
【问题描述】：

我有一个简单的自动加载器作业，看起来像这样

df_dwu_limit = spark.readStream.format("cloudFiles") \
    .option("cloudFiles.format", "JSON") \
    .schema(schemaFromJson) \
    .load("abfss://synapse-usage@xxxxx.dfs.core.windows.net/synapse-usage/")\
    .writeStream \
    .format("delta")\
    .option("checkpointLocation", "abfss://adb-delta-synapse-usage-api@xxxxxx.dfs.core.windows.net/checkpoint_synapse_usage_api_landing/") \
    .trigger(availableNow=True)\
    .toTable("platform_dnu.synapse_usage_api_landing")

下一行我正在使用增量表“”进行进一步处理。

我在自动加载器之前和之后也有 count(*) 查询。计数不变。虽然我可以在自动装载机配置文件中看到，记录已经写入。

如果我等待 1 分钟左右，然后运行 count(*) 查询，我可以看到更新的记录。如何解决这个问题？

一个特定会话的自动加载器的 o/p

{
  "id" : "cb9a28b4-c5b4-4865-bc65-b3ca5efd2537",
  "runId" : "64c2afd9-ad69-4e9a-97bf-d6fa2794931a",
  "name" : null,
  "timestamp" : "2022-12-03T04:44:17.591Z",
  "batchId" : 7,
  "numInputRows" : 27,
  "inputRowsPerSecond" : 0.0,
  "processedRowsPerSecond" : 0.7879760688749453,
  "durationMs" : {
    "addBatch" : 3005,
    "commitOffsets" : 146,
    "getBatch" : 12,
    "latestOffset" : 30380,
    "queryPlanning" : 61,
    "triggerExecution" : 34259,
    "walCommit" : 222
  },
  "stateOperators" : [ ],
  "sources" : [ {
    "description" : "CloudFilesSource[abfss://synapse-usage@xxxxx.dfs.core.windows.net/synapse-usage/]",
    "startOffset" : {
      "seqNum" : 2534,
      "sourceVersion" : 1,
      "lastBackfillStartTimeMs" : 1669823987701,
      "lastBackfillFinishTimeMs" : 1669823991340
    },
    "endOffset" : {
      "seqNum" : 2562,
      "sourceVersion" : 1,
      "lastBackfillStartTimeMs" : 1669823987701,
      "lastBackfillFinishTimeMs" : 1669823991340
    },
    "latestOffset" : null,
    "numInputRows" : 27,
    "inputRowsPerSecond" : 0.0,
    "processedRowsPerSecond" : 0.7879760688749453,
    "metrics" : {
      "numBytesOutstanding" : "0",
      "numFilesOutstanding" : "0"
    }
  } ],
  "sink" : {
    "description" : "DeltaSink[abfss://adb-delta-synapse-usage-api@xxxx.dfs.core.windows.net/delta/synapse_usage_api_landing]",
    "numOutputRows" : -1
  }}

增量表的 DDL。

【问题讨论】：

标签： databricks azure-databricks

【解决方案1】：

在自动加载器解决问题后放置以下代码。

df_dwu_limit.awaitTermination()

【讨论】：