私のスパークストリーミングアプリケーションが停止/終了した場合に、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のprebuild sparkとspark sourceとともに提供されるspark-examplesには多くの例があります。具体的には、JavaStatefulNetworkWordCount.javaを参照してください。updateStateByKeyreduceByKeyAndWindow
この記事はインターネットから収集されたものであり、転載の際にはソースを示してください。
侵害の場合は、連絡してください[email protected]
コメントを追加