我想用spark文件流应用程序实现检查点,以在任何情况下我的spark流应用程序停止/终止时处理hadoop中所有未处理的文件。我正在遵循:流式编程指南,但未找到JavaStreamingContextFactory。请帮我该怎么办。
我的代码是
public class StartAppWithCheckPoint {
public static void main(String[] args) {
try {
String filePath = "hdfs://Master:9000/mmi_traffic/listenerTransaction/2020/*/*/*/";
String checkpointDirectory = "hdfs://Mongo1:9000/probeAnalysis/checkpoint";
SparkSession sparkSession = JavaSparkSessionSingleton.getInstance();
JavaStreamingContextFactory contextFactory = new JavaStreamingContextFactory() {
@Override public JavaStreamingContext create() {
SparkConf sparkConf = new SparkConf().setAppName("ProbeAnalysis");
JavaSparkContext sc = new JavaSparkContext(sparkConf);
JavaStreamingContext jssc = new JavaStreamingContext(sc, Durations.seconds(300));
JavaDStream<String> lines = jssc.textFileStream(filePath).cache();
jssc.checkpoint(checkpointDirectory);
return jssc;
}
};
JavaStreamingContext context = JavaStreamingContext.getOrCreate(checkpointDirectory, contextFactory);
context.start();
context.awaitTermination();
context.close();
sparkSession.close();
} catch(Exception e) {
e.printStackTrace();
}
}
}
您必须使用检查点
对于检查点,请使用或有状态转换。在火花示例中有很多示例,以及git-hub中的预构建火花和火花源。有关您的具体信息,请参见JavaStatefulNetworkWordCount.java。updateStateByKey
reduceByKeyAndWindow
本文收集自互联网,转载请注明来源。
如有侵权,请联系 [email protected] 删除。
我来说两句