时间戳

首页 > 解决方案 > 引发流无法写入 hdfs 路径

apache-spark - 引发流无法写入 hdfs 路径

问题描述

我使用 java 1.8 和 kafka 0.10.x 一起使用 spark-sql-2.4.1v。

Dataset<Row> dataSet= sparkSession
                      .readStream()
                      .format("kafka")
                      .option("subscribe", INFO_TOPIC)
                      .option("startingOffsets", "latest")
                      .option("enable.auto.commit", false)
                      .option("maxOffsetsPerTrigger", 1000)
                      .option("auto.offset.reset", "latest")
                      .option("failOnDataLoss", false)
                      .load();



StreamingQuery query = dataSet.writeStream()
        .format(PARQUET_FORMAT)
        .option("path", parqetFileName)
        .option("checkpointLocation", checkPtLocation)
        .trigger(Trigger.ProcessingTime("15 seconds"))
        .start();



query.awaitTermination();

将数据写入我的 hdfs 路径（即 parqetFileName）后，它失败并出现以下错误。

[DataStreamer for file /user/parquet/raw/part-00001-7cba7fa3-a98f-442d-9584-b71085b7cd82-c000.snappy.parquet] WARN  org.apache.hadoop.hdfs.DataStreamer - Caught exception
java.lang.InterruptedException
        at java.lang.Object.wait(Native Method)
        at java.lang.Thread.join(Thread.java:1249)
        at java.lang.Thread.join(Thread.java:1323)
        at org.apache.hadoop.hdfs.DataStreamer.closeResponder(DataStreamer.java:980)
        at org.apache.hadoop.hdfs.DataStreamer.endBlock(DataStreamer.java:630)
        at org.apache.hadoop.hdfs.DataStreamer.run(DataStreamer.java:807)

这里有什么问题以及如何解决？

标签： apache-sparkhadoopapache-spark-sqlhdfsspark-structured-streaming

解决方案

您的代码中必须包含streamContext.awaitTermination()- 否则应用程序将在启动流后立即退出。

推荐阅读