0

我已经使用 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 我的代码中是否缺少某些内容？如何启动流式处理并将数据集写入文件系统？任何帮助表示赞赏，谢谢。

4

0 回答 0