Apache Spark、带有自定义接收器的 Java 流挂起

编程问答 2022-05-12

问题描述

我已经使用 java 流 API 编写了一个自定义接收器。 它的目的是读取加密文件，对其进行解密，然后推送行以进行进一步详细说明。 问题是，即使文件被正确读取、解密并推送到内存存储，我也无法开始流式传输。 使流生成输出的唯一方法是优雅地停止火花流上下文。

这是我的 java 自定义接收器的一部分，我将解密的行放入商店：

BufferedReader br = new BufferedReader(new InputStreamReader(zsIn,StandardCharsets.UTF_8));
for (int i = 0; !this.isstopped() && (decompressedLine = br.readLine()) != null; ++i) {
                this.store(decompressedLine);
}

这里是我访问接收到的流并尝试执行一些操作的地方：

SparkConf sparkConf = new SparkConf().setAppName("MyJavaStreaming");
JavaStreamingContext ssc = new JavaStreamingContext(sparkConf,Durations.minutes(1));
JavaReceiverInputDStream<String> customreceiverStream = ssc.receiverStream(new SparkDecodingReceiver(args[0],args[1]));
customreceiverStream.foreachRDD((rdd,time) -> {
    SparkSession singletonSpark = JavaSparkSessionSingleton.getInstance(rdd.context().getConf());
    JavaRDD<InputFileFlatStructure> rowRDD = rdd.map(line -> {
        InputFileFlatStructure inputRecord = new InputFileFlatStructure();
        if (!line.isEmpty()) {
            inputRecord.setValue(line);
        }
        return inputRecord;
    });
    
    Dataset<Row> wordsDataFrame = singletonSpark.createDataFrame(rowRDD,InputFileFlatStructure.class);
    wordsDataFrame.coalesce(1).write().format("csv").option("sep","|").save("/nfs/output");

});

ssc.start();
ssc.awaitTermination();

使用此代码，进程会挂起并且永远不会刷新文件系统上接收到的流。 除了自定义接收器和火花存储的使用之外，代码与此示例非常相似： https://github.com/apache/spark/blob/v3.1.1/examples/src/main/java/org/apache/spark/examples/streaming/JavaSqlNetworkWordCount.java 我的代码中是否缺少某些内容？如何启动流处理并将数据集写入文件系统？ 感谢任何帮助，谢谢。

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）

apache-sparkcustom-receiverjavajavaspark-streaming