结构化流写入Azure突触

支持输出流写入突触模式

Azure突触连接器支持附加和完整的输出模式记录附加和聚合。为更多的细节在输出模式和兼容性矩阵,看到结构化流导。

突触容错语义

默认情况下,Azure突触流提供了端到端仅一次保证写数据到Azure突触表可靠地跟踪进度查询的使用检查点位置DBFS、检查点表在Azure突触,和锁定机制,以确保流可以处理任何类型的失败,重试,和查询将重新启动。

可选地,您可以选择限制较少“至少一次”语义Azure突触流设置spark.databricks.sqldw.streaming.exactlyOnce.enabled选项假,在这种情况下,数据重复的事件可能发生间歇性连接失败Azure突触查询或意外终止。

结构化流语法写作Azure突触

下面的代码示例演示流写入突触在Scala和Python使用结构化流:

/ /设置Blob存储帐户访问关键会话配置在笔记本上。火花。相依。集(“fs.azure.account.key < your-storage-account-name > .dfs.core.windows.net”。,“< your-storage-account-access-key >”)/ /准备流源;这可能是卡夫卡或一个简单的流率。瓦尔df:DataFrame=火花。readStream。格式(“速度”)。选项(“rowsPerSecond”,“100000”)。选项(“numPartitions”,“16”)。负载()/ /应用一些转换到数据然后使用/ /结构化流API,不断将数据写入一个表在Azure突触。df。writeStream。格式(“com.databricks.spark.sqldw”)。选项(“url”," jdbc::状态"置疑" / / < the-rest-of-the-connection-string >”)。选项(“tempDir”,“abfss: / / < your-container-name > @ < your-storage-account-name >.dfs.core.windows.net/ < your-directory-name >”)。选项(“forwardSparkAzureStorageCredentials”,“真正的”)。选项(“数据表”,“< your-table-name >”)。选项(“checkpointLocation”,“/ tmp_checkpoint_location”)。开始()

#设置Blob存储帐户访问关键会话配置在笔记本上。火花。相依。集(“fs.azure.account.key < your-storage-account-name > .dfs.core.windows.net”。,“< your-storage-account-access-key >”)#准备流源;这可能是卡夫卡或一个简单的流率。df=火花。readStream\。格式(“速度”)\。选项(“rowsPerSecond”,“100000”)\。选项(“numPartitions”,“16”)\。负载()然后使用#应用一些转换数据#结构化流API,不断用Azure突触写数据表。df。writeStream\。格式(“com.databricks.spark.sqldw”)\。选项(“url”," jdbc::状态"置疑" / / < the-rest-of-the-connection-string >”)\。选项(“tempDir”,“abfss: / / < your-container-name > @ < your-storage-account-name >.dfs.core.windows.net/ < your-directory-name >”)\。选项(“forwardSparkAzureStorageCredentials”,“真正的”)\。选项(“数据表”,“< your-table-name >”)\。选项(“checkpointLocation”,“/ tmp_checkpoint_location”)\。开始()

配置的完整列表,请参阅在Azure突触分析查询数据。

突触流表管理检查点

Azure突触连接器不删除流检查点表时创建新的流媒体查询开始。这种行为是一致的checkpointLocation通常指定对象存储。砖建议您定期删除检查点的表进行查询,在未来不会运行。

默认情况下,所有检查点表名称<前缀> _ < query-id >,在那里<前缀>是一个可配置的前缀与默认值databricks_streaming_checkpoint和query_id是一个流媒体查询ID_角色删除。

寻找过期或删除所有检查点表流查询,运行查询:

选择*从sys。表在哪里的名字就像“databricks_streaming_checkpoint %”

您可以配置前缀与火花SQL配置选项spark.databricks.sqldw.streaming.exactlyOnce.checkpointTableNamePrefix。

砖突触连接器流选择参考

的选项火花SQL支持以下选项中提供流媒体除了批处理选项: