无法使用Spark Structured Streaming在Parquet文件中写入数据

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了无法使用Spark Structured Streaming在Parquet文件中写入数据相关的知识,希望对你有一定的参考价值。

我有一个Spark结构化流:

val df = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "localhost:9092")
      .option("startingOffsets", "earliest")
      .option("endingOffsets", "latest")
      .option("subscribe", "topic")
      .load()

我想使用DataStreamWriter将数据写入FileSystem,

val query = df
          .writeStream
          .outputMode("append")
          .format("parquet")
          .start("data")

但是在data文件夹中创建了零文件。只有_spark_metadata正在创建。

但是,当formatconsole时,我可以在控制台上看到数据:

val query = df
          .writeStream
          .outputMode("append")
          .format("console")
          .start()

+--------------------+------------------+------------------+
|                time|              col1|              col2|
+--------------------+------------------+------------------+
|49368-05-11 20:42...|0.9166470338147503|0.5576946794171861|
+--------------------+------------------+------------------+

我无法理解它背后的原因。

Spark - 2.1.0

答案

我解决了这个问题。实际上,当我尝试在spark-shell上运行Structured Streaming时,它给出了一个错误,即endingOffsets在流式查询中无效,即:

val df = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "localhost:9092")
      .option("startingOffsets", "earliest")
      .option("endingOffsets", "latest")
      .option("subscribe", "topic")
      .load()


java.lang.IllegalArgumentException: ending offset not valid in streaming queries
  at org.apache.spark.sql.kafka010.KafkaSourceProvider$$anonfun$validateStreamOptions$1.apply(KafkaSourceProvider.scala:374)
  at org.apache.spark.sql.kafka010.KafkaSourceProvider$$anonfun$validateStreamOptions$1.apply(KafkaSourceProvider.scala:373)
  at scala.Option.map(Option.scala:146)
  at org.apache.spark.sql.kafka010.KafkaSourceProvider.validateStreamOptions(KafkaSourceProvider.scala:373)
  at org.apache.spark.sql.kafka010.KafkaSourceProvider.sourceSchema(KafkaSourceProvider.scala:60)
  at org.apache.spark.sql.execution.datasources.DataSource.sourceSchema(DataSource.scala:199)
  at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo$lzycompute(DataSource.scala:87)
  at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo(DataSource.scala:87)
  at org.apache.spark.sql.execution.streaming.StreamingRelation$.apply(StreamingRelation.scala:30)
  at org.apache.spark.sql.streaming.DataStreamReader.load(DataStreamReader.scala:124)
  ... 48 elided

所以,我从流式查询中删除了endingOffsets

val df = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "localhost:9092")
      .option("startingOffsets", "earliest")
      .option("subscribe", "topic")
      .load()

然后我尝试在Parquet文件中保存流式查询的结果,在此期间我开始知道 - 必须指定检查点位置,即:

val query = df
          .writeStream
          .outputMode("append")
          .format("parquet")
          .start("data")

org.apache.spark.sql.AnalysisException: checkpointLocation must be specified either through option("checkpointLocation", ...) or SparkSession.conf.set("spark.sql.streaming.checkpointLocation", ...);
  at org.apache.spark.sql.streaming.StreamingQueryManager$$anonfun$3.apply(StreamingQueryManager.scala:207)
  at org.apache.spark.sql.streaming.StreamingQueryManager$$anonfun$3.apply(StreamingQueryManager.scala:204)
  at scala.Option.getOrElse(Option.scala:121)
  at org.apache.spark.sql.streaming.StreamingQueryManager.createQuery(StreamingQueryManager.scala:203)
  at org.apache.spark.sql.streaming.StreamingQueryManager.startQuery(StreamingQueryManager.scala:269)
  at org.apache.spark.sql.streaming.DataStreamWriter.start(DataStreamWriter.scala:262)
  at org.apache.spark.sql.streaming.DataStreamWriter.start(DataStreamWriter.scala:206)
  ... 48 elided

所以,我添加了checkPointLocation

val query = df
          .writeStream
          .outputMode("append")
          .format("parquet")
          .option("checkpointLocation", "checkpoint")
          .start("data")

完成这些修改后,我能够在Parquet文件中保存流式查询的结果。

但是,奇怪的是,当我通过sbt应用程序运行相同的代码时,它没有抛出任何错误,但是当我通过spark-shell运行相同的代码时,这些错误被抛出。我认为Apache Spark也会在通过sbt / maven app运行时抛出这些错误。这对我来说似乎是个错误!

另一答案

我有类似的问题,但出于不同的原因,发布在这里以防有人有同样的问题。当使用水印将输出流写入附加模式的文件时,结构化流式传输具有一种有趣的行为,即在时间桶早于水印时间之前它不会实际写入任何数据。如果您正在测试结构化流媒体并且有一个小时的水印,那么至少一小时内您将看不到任何输出。

以上是关于无法使用Spark Structured Streaming在Parquet文件中写入数据的主要内容,如果未能解决你的问题,请参考以下文章

如何将 from_json 与 Kafka connect 0.10 和 Spark Structured Streaming 一起使用?

Spark Structured Streaming - 此查询不支持从检查点位置恢复

Spark Structured Streaming

Spark Structured Streaming

如何使用Spark Structured Streaming连续监视目录

Spark Streaming:Spark Structured Streaming 中不允许使用 Kafka 组 ID