使用结构化流式处理写入 Azure Synapse

Azure Synapse 连接器为 Azure Synapse 提供高效且可缩放的结构化流式写入支持，通过提供一致的用户体验和批量写入功能，并使用 COPY 在 Azure Databricks 群集与 Azure Synapse 实例之间进行大规模数据传输。

Azure Databricks 与 Synapse 之间的结构化流式处理支持为配置增量 ETL 作业提供简单的语义。 用于将 Azure Databricks 中的数据加载到 Synapse 的模型会造成可能不符合准实时工作负载 SLA 要求的延迟。 请参阅在 Azure Synapse Analytics 中查询数据。

支持流式写入到 Synapse 的输出模式

Azure Synapse 连接器支持用于记录追加和聚合的 Append 和 Complete 输出模式。 有关输出模式和兼容性矩阵的更多详细信息，请参阅结构化流式处理指南。

Synapse 容错语义

默认情况下，Azure Synapse 流式处理提供端到端的精确一次性保证，通过可靠地跟踪查询进度，将 DBFS 中的检查点位置、Azure Synapse 中的检查点表以及锁定机制组合使用，确保流式处理可以应对任何类型的故障、重试和查询重启，从而实现将数据写入 Azure Synapse 表。

可以通过将 spark.databricks.sqldw.streaming.exactlyOnce.enabled 选项设置为 false，为 Azure Synapse 流式处理选择限制较少的“至少一次”语义。在这种情况下，如果连接到 Azure Synapse 时发生间歇性故障或查询意外终止，可能会导致数据重复出现。

用于将数据写入 Azure Synapse 的结构化流处理语法

以下代码示例演示如何通过在 Scala 和 Python 中使用结构化流进行流写入到 Synapse：

Scala(编程语言)

// Set up the Blob storage account access key in the notebook session conf.
spark.conf.set(
  "fs.azure.account.key.<your-storage-account-name>.dfs.core.chinacloudapi.cn",
  "<your-storage-account-access-key>")

// Prepare streaming source; this could be Kafka or a simple rate stream.
val df: DataFrame = spark.readStream
  .format("rate")
  .option("rowsPerSecond", "100000")
  .option("numPartitions", "16")
  .load()

// Apply some transformations to the data then use
// Structured Streaming API to continuously write the data to a table in Azure Synapse.

df.writeStream
  .format("com.databricks.spark.sqldw")
  .option("url", "jdbc:sqlserver://<the-rest-of-the-connection-string>")
  .option("tempDir", "abfss://<your-container-name>@<your-storage-account-name>.dfs.core.chinacloudapi.cn/<your-directory-name>")
  .option("forwardSparkAzureStorageCredentials", "true")
  .option("dbTable", "<your-table-name>")
  .option("checkpointLocation", "/tmp_checkpoint_location")
  .start()

Python

# Set up the Blob storage account access key in the notebook session conf.
spark.conf.set(
  "fs.azure.account.key.<your-storage-account-name>.dfs.core.chinacloudapi.cn",
  "<your-storage-account-access-key>")

# Prepare streaming source; this could be Kafka or a simple rate stream.
df = spark.readStream \
  .format("rate") \
  .option("rowsPerSecond", "100000") \
  .option("numPartitions", "16") \
  .load()

# Apply some transformations to the data then use
# Structured Streaming API to continuously write the data to a table in Azure Synapse.

df.writeStream \
  .format("com.databricks.spark.sqldw") \
  .option("url", "jdbc:sqlserver://<the-rest-of-the-connection-string>") \
  .option("tempDir", "abfss://<your-container-name>@<your-storage-account-name>.dfs.core.chinacloudapi.cn/<your-directory-name>") \
  .option("forwardSparkAzureStorageCredentials", "true") \
  .option("dbTable", "<your-table-name>") \
  .option("checkpointLocation", "/tmp_checkpoint_location") \
  .start()

有关完整的配置列表，请参阅在 Azure Synapse Analytics 中查询数据。

Synapse 流式处理检查点表管理

Azure Synapse 连接器不会删除在新的流式查询启动时创建的流式处理检查点表。 此行为与通常为对象存储指定的 checkpointLocation 相一致。 Databricks 建议定期删除将来不会运行的查询的检查点表。

默认情况下，所有检查点表的名称都为 <prefix>_<query-id>，其中 <prefix> 是一个可配置前缀（默认值为 databricks_streaming_checkpoint），query_id 是删除了 _ 字符的流式查询 ID。

若要查找陈旧的或已删除的流式查询的所有检查点表，请运行以下查询：

SELECT * FROM sys.tables WHERE name LIKE 'databricks_streaming_checkpoint%'

可以通过 Spark SQL 配置选项 spark.databricks.sqldw.streaming.exactlyOnce.checkpointTableNamePrefix 来配置前缀。

Azure Databricks Synapse 连接器流式处理选项参考

在 Spark SQL 中提供的 OPTIONS 中，除了 批处理选项 外，还支持以下流式处理选项：