Databricks 无法保存流检查点答案

【问题标题】：Databricks cannot save stream checkpointDatabricks 无法保存流检查点
【发布时间】：2022-01-16 20:22:54
【问题描述】：

我正在尝试设置流以开始处理传入文件。看起来 Databricks 无法保存检查点。我尝试了 ADLS Gen2 和 DBFS 中的位置，结果相同。 Databricks 使用某些结构创建所需的文件夹，但无法写入。检查点位置有什么特殊要求吗？

Checkpoint folder

Databricks 社区版，运行时版本：9.1 LTS（包括 Apache Spark 3.1.2、Scala 2.12）

spark.readStream
.format("cloudFiles")
.option("cloudFiles.format", "parquet")
.option("cloudFiles.partitionColumns", "year, month, day")
.option("header", "true")
.schema(schema)
.load(destFolderName)
.writeStream.format("delta")
.option("checkpointLocation", checkpointPath)
.outputMode("append")
.partitionBy("year", "month", "day")
.start(outputPath)

错误：

java.lang.UnsupportedOperationException: com.databricks.backend.daemon.data.client.DBFSV1.createAtomicIfAbsent(path: Path)
at com.databricks.tahoe.store.EnhancedDatabricksFileSystemV1.createAtomicIfAbsent(EnhancedFileSystem.scala:324)
at com.databricks.spark.sql.streaming.AWSCheckpointFileManager.createAtomicIfAbsent(DatabricksCheckpointFileManager.scala:159)
at com.databricks.spark.sql.streaming.DatabricksCheckpointFileManager.createAtomicIfAbsent(DatabricksCheckpointFileManager.scala:60)
at com.databricks.sql.streaming.state.RocksDBFileManager.zipToDbfsFile(RocksDBFileManager.scala:497)
at com.databricks.sql.streaming.state.RocksDBFileManager.saveCheckpointToDbfs(RocksDBFileManager.scala:181)
at com.databricks.sql.rocksdb.CloudRocksDB.$anonfun$open$5(CloudRocksDB.scala:451)
at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
at org.apache.spark.util.Utils$.timeTakenMs(Utils.scala:627)
at com.databricks.sql.rocksdb.CloudRocksDB.timeTakenMs(CloudRocksDB.scala:527)
at com.databricks.sql.rocksdb.CloudRocksDB.$anonfun$open$2(CloudRocksDB.scala:439)
at com.databricks.logging.UsageLogging.$anonfun$recordOperation$1(UsageLogging.scala:395)
at com.databricks.logging.UsageLogging.executeThunkAndCaptureResultTags$1(UsageLogging.scala:484)
at com.databricks.logging.UsageLogging.$anonfun$recordOperationWithResultTags$4(UsageLogging.scala:504)
at com.databricks.logging.UsageLogging.$anonfun$withAttributionContext$1(UsageLogging.scala:266)
at scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)
at com.databricks.logging.UsageLogging.withAttributionContext(UsageLogging.scala:261)
at com.databricks.logging.UsageLogging.withAttributionContext$(UsageLogging.scala:258)
at com.databricks.spark.util.PublicDBLogging.withAttributionContext(DatabricksSparkUsageLogger.scala:20)

【问题讨论】：

标签： spark-streaming databricks azure-databricks databricks-community-edition databricks-autoloader

【解决方案1】：

您可以尝试禁用多集群写入：

spark.databricks.delta.multiClusterWrites.enabled false

检查您的路径 - 请尝试写入由数据块管理的标准 dbfs（例如 dbfs:/local_disk0/tmp/checkpointName ）。

如果您使用自己的挂载，请检查那里的 azure 权限（需要 Blob 存储贡献者）。

请诊断同时读取流

df = spark.readStream(...)
display(df)

【讨论】：

感谢您的回复。我尝试禁用多集群写入。我可以成功地读取和写入镶木地板文件到 DBFS 和 ADLS Gen2，但不能在流模式下。已分配 Blob 存储参与者。我注释掉了写作部分，现在阅读失败了，我猜在“load（）”行。这个问题是否与 Databricks Community Edition 在 DBFSV1 上运行的事实有关？ kb.databricks.com/delta/delta-write-fails.html
可以，但我不知道有DBFSV1。一般来说社区版是有限的，我只做简单的工作人员，例如测试 sql 查询。我还没有在那里流式传输。

【解决方案2】：

我尝试使用的自动加载器功能目前在 Databricks 社区版上不可用

https://databricks.com/notebooks/Databricks-Data-Integration-Demo.html

所以“cloudFiles”不能与社区版一起使用

【讨论】：