현재값과 다음값 비교 자바 버전 compare current value & next value java version

// 인덱스 부여 JavaPairRDD<String, Long> lineIndexRDD = jsc.textFile(path).persist(storageLevel).zipWithIndex(); // key, value pair 로 변환 (key: index, value: line) index 는 0 부터 시작 JavaPairRDD<Long, String> currLine = lineIndexRDD.mapToPair(new PairFunction<Tuple2<String, Long>, Long, String>() { @Override public Tuple2<Long, String> call(Tuple2<String, Long> lineIndex) throws Exception { return new Tuple2<>(lineIndex._2(), lineIndex._1()); } }); // 다음값 추출 JavaPairRDD<Long, String> nextLine = lineIndexRDD.filter(new Function<Tuple2<String, Long>, Boolean>() { @Override public Boolean call(Tuple2<String, Long> v1) throws Exception { return v1._2() > 0; } }).mapToPair(new PairFunction<Tuple2<String, Long>, Long, String>() { @Override public Tuple2<Long, String> call(Tuple2<String, Long> lineIndex) throws Exception { return new Tuple2<>(lineIndex._2() - 1, lineIndex._1()); } }); // 부여한 index 로 join JavaRDD<String> compareResult = currLine.join(nextLine).map(new Function<Tuple2<Long, Tuple2<String, String>>, String>() { @Override public String call(Tuple2<Long, Tuple2<String, String>> indexCurrNext) throws Exception { String curr = indexCurrNext._2()._1(); String next = indexCurrNext._2()._2(); // your compare logic return ""; } });

제외되는 값 없이 비교하기 (비교 대상이 없는 경우 다음 값 반환)

// 인덱스 부여 JavaPairRDD<String, Long> lineIndexRDD = jsc.textFile("").persist(null).zipWithIndex(); // key, value pair 로 변환 (key: index, value: line) index 는 0 부터 시작 JavaPairRDD<Long, String> currLine = lineIndexRDD.mapToPair(new PairFunction<Tuple2<String, Long>, Long, String>() { @Override public Tuple2<Long, String> call(Tuple2<String, Long> lineIndex) throws Exception { return new Tuple2<>(lineIndex._2(), lineIndex._1()); } }); // 다음값 추출 JavaPairRDD<Long, String> nextLine = lineIndexRDD.mapToPair(new PairFunction<Tuple2<String, Long>, Long, String>() { @Override public Tuple2<Long, String> call(Tuple2<String, Long> lineIndex) throws Exception { return new Tuple2<>(lineIndex._2() + 1, lineIndex._1()); } }); // 부여한 index 로 join JavaRDD<String> compareResult = currLine.leftOuterJoin(nextLine).map(new Function<Tuple2<Long, Tuple2<String, Optional<String>>>, String>() { @Override public String call(Tuple2<Long, Tuple2<String, Optional<String>>> indexCurrNext) throws Exception { String next = indexCurrNext._2()._1(); Optional<String> currOptional = indexCurrNext._2()._2(); String curr = null; if (currOptional.isPresent()) { curr = currOptional.get(); } else { return next; } // your compare logic return next; } });