Checkpoint dengan file spark streaming di java

Question

Saya ingin menerapkan checkpoint dengan aplikasi streaming file percikan untuk memproses semua file yang belum diproses dari hadoop jika dalam hal apa pun aplikasi streaming percikan saya berhenti / dihentikan. Saya mengikuti ini: panduan pemrograman streaming , tetapi tidak ditemukan JavaStreamingContextFactory. Tolong bantu saya apa yang harus saya lakukan.

Kode Saya adalah

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();
        }   
    }
}

Answer 1

Anda harus menggunakan Checkpointing

Untuk checkpointing gunakan transformasi stateful baik updateStateByKeyatau reduceByKeyAndWindow. Ada banyak contoh dalam contoh percikan yang disediakan bersama dengan percikan prebuild dan sumber percikan di git-hub. Untuk spesifik Anda, lihat JavaStatefulNetworkWordCount.java ;