为什么流式查询不向 HDFS 写入任何数据？答案

【问题标题】：Why does streaming query not write any data to HDFS?为什么流式查询不向 HDFS 写入任何数据？
【发布时间】：2018-12-24 16:58:53
【问题描述】：

我正在使用带有 Spark 2.3.1 的 Spark Structured Streaming，以下是我的代码：

val sparkSession = SparkSession
.builder
.appName("xxx")
.config("spark.serializer", 
  "org.apache.spark.serializer.KryoSerializer")
.config("spark.rpc.netty.dispatcher.numThreads", "2")
.config("spark.shuffle.compress", "true")
.config("spark.rdd.compress", "true")
.config("spark.sql.inMemoryColumnarStorage.compressed", "true")
.config("spark.io.compression.codec", "snappy")
.config("spark.broadcast.compress", "true")
.config("spark.sql.hive.thriftServer.singleSession", "true")
.config("hive.exec.dynamic.partition", "true")
.config("hive.exec.dynamic.partition.mode", "nonstrict")
.config("spark.streaming.receiver.writeAheadLog.enable","true")
.enableHiveSupport()
.getOrCreate()

val rawStreamDF = sparkSession
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", <value>)
.option("subscribe", <value>)
.option("key.serializer", <value>)
.option("value.serializer", <value>)
.option("startingOffsets", "earliest")
.option("auto.offset.reset",earliest)
.option("group.id",  <value>)
.option("fetchOffset.numRetries", 3)
.option("fetchOffset.retryIntervalMs", 10)
.option("IncludeTimestamp", true)
.option("enable.auto.commit",  <value>)
.option("security.protocol",  <value>)
.option("ssl.keystore.location",  <value>)
.option("ssl.keystore.password",  <value>)
.option("ssl.truststore.location",  <value>)
.option("ssl.truststore.password",  <value>)
.load()
.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
.as[(String, String)]

我正在尝试将数据写入 hdfs_path 中的文件：

val query = rawStreamDF
  .writeStream
  .format("json")
  .option("startingOffsets", "latest")
  .option("path", "STREAM_DATA_PATH")
  .option("checkpointLocation", "checkpointPath")
  .trigger(Trigger.ProcessingTime("5 seconds"))
  .outputMode("append")
  .start

Logger.log.info("Status:"+query.status)
print("Streaming Status1:"+query.status)

query.awaitTermination(450)

但是，我得到的 query.status 值如下：

Status:{ "message" : "Initializing sources", "isDataAvailable" : false, "isTriggerActive" : false }

你能告诉我哪里出错了吗？

【问题讨论】：

标签： scala apache-spark spark-structured-streaming

【解决方案1】：

但是，我得到的 query.status 值如下。
Status:{ "message" : "Initializing sources", "isDataAvailable" :false, "isTriggerActive" : false }
你能告诉我哪里出错了吗？

一切似乎都很好。 Spark Structured Streaming 的流引擎似乎还没有启动查询，只是将其标记为在单独的线程上启动。

如果您创建了一个单独的线程来监控结构化查询，您会注意到在处理完第一批后状态会立即发生变化。

请参阅Structured Streaming Programming Guide中的官方文档。

【讨论】：