@Test public void savePairRDDShouldSavePairRDDAsSeqFiles() throws IOException { JavaPairRDD<Text, Text> in = sc.parallelizePairs(Arrays.asList( new Tuple2<>(new Text("1L"), new Text("1R")), new Tuple2<>(new Text("2L"), new Text("2R")), new Tuple2<>(new Text("3L"), new Text("3R"))), NUMBER_OF_OUTPUT_FILES + 1); RDDUtils.saveTextPairRDD(in, NUMBER_OF_OUTPUT_FILES, outputDir.toString(), configuration); Pattern pattern = Pattern.compile("^part-r-.\\d+$"); long fileCount = HdfsUtils.countFiles(new Configuration(), outputDir.toString(), x -> pattern.matcher(x.getName()).matches()); assertEquals(NUMBER_OF_OUTPUT_FILES, fileCount); List<Text> out = ListTestUtils.readValues(outputDir.toString(), Function.identity()); List<String> actualValues = out.stream().map(Text::toString).sorted().collect(Collectors.toList()); List<String> expectedValues = in.collect().stream().map(x -> x._2.toString()).sorted().collect(Collectors.toList()); ListTestUtils .compareLists(actualValues, expectedValues); }
public static void saveTextPairRDD(JavaPairRDD<Text, Text> pairs, Integer numberOfOutputFiles, String outputPath, Configuration conf) { pairs .repartition(numberOfOutputFiles) .saveAsNewAPIHadoopFile(outputPath, Text.class, Text.class, SequenceFileOutputFormat.class, conf); }
RDDUtils { public static void saveTextPairRDD(JavaPairRDD<Text, Text> pairs, Integer numberOfOutputFiles, String outputPath, Configuration conf) { pairs .repartition(numberOfOutputFiles) .saveAsNewAPIHadoopFile(outputPath, Text.class, Text.class, SequenceFileOutputFormat.class, conf); } }
RDDUtils { public static void saveTextPairRDD(JavaPairRDD<Text, Text> pairs, Integer numberOfOutputFiles, String outputPath, Configuration conf) { pairs .repartition(numberOfOutputFiles) .saveAsNewAPIHadoopFile(outputPath, Text.class, Text.class, SequenceFileOutputFormat.class, conf); } private  RDDUtils(); }
RDDUtils { public static void saveTextPairRDD(JavaPairRDD<Text, Text> pairs, Integer numberOfOutputFiles, String outputPath, Configuration conf) { pairs .repartition(numberOfOutputFiles) .saveAsNewAPIHadoopFile(outputPath, Text.class, Text.class, SequenceFileOutputFormat.class, conf); } private  RDDUtils(); static void saveTextPairRDD(JavaPairRDD<Text, Text> pairs, Integer numberOfOutputFiles, String outputPath, Configuration conf); }
RDDUtils { public static void saveTextPairRDD(JavaPairRDD<Text, Text> pairs, Integer numberOfOutputFiles, String outputPath, Configuration conf) { pairs .repartition(numberOfOutputFiles) .saveAsNewAPIHadoopFile(outputPath, Text.class, Text.class, SequenceFileOutputFormat.class, conf); } private  RDDUtils(); static void saveTextPairRDD(JavaPairRDD<Text, Text> pairs, Integer numberOfOutputFiles, String outputPath, Configuration conf);  }