@Test public void testRRMDSIJoin() throws Exception { ClassPathResource cpr1 = new ClassPathResource("spark/rrmdsi/file1.txt"); ClassPathResource cpr2 = new ClassPathResource("spark/rrmdsi/file2.txt"); RecordReader rr1 = new CSVRecordReader(); rr1.initialize(new FileSplit(cpr1.getFile())); RecordReader rr2 = new CSVRecordReader(); rr2.initialize(new FileSplit(cpr2.getFile())); RecordReaderMultiDataSetIterator rrmdsi1 = new RecordReaderMultiDataSetIterator.Builder(1) .addReader("r1", rr1) .addReader("r2", rr2) .addInput("r1", 1, 2) .addOutput("r2",1,2) .build(); RecordReaderMultiDataSetIterator rrmdsi2 = new RecordReaderMultiDataSetIterator.Builder(1) .addReader("r1", new SparkSourceDummyReader(0)) .addReader("r2", new SparkSourceDummyReader(1)) .addInput("r1", 1, 2) .addOutput("r2",1,2) .build(); List<MultiDataSet> expected = new ArrayList<>(3); while(rrmdsi1.hasNext()){ expected.add(rrmdsi1.next()); } JavaRDD<List<Writable>> rdd1 = sc.textFile(cpr1.getFile().getPath()).coalesce(1) .map(new StringToWritablesFunction(new CSVRecordReader())); JavaRDD<List<Writable>> rdd2 = sc.textFile(cpr2.getFile().getPath()).coalesce(1) .map(new StringToWritablesFunction(new CSVRecordReader())); List<JavaRDD<List<Writable>>> list = Arrays.asList(rdd1, rdd2); JavaRDD<MultiDataSet> mdsRdd = IteratorUtils.mapRRMDSI(list, null, new int[]{0,0}, null, false, rrmdsi2); List<MultiDataSet> act = mdsRdd.collect(); expected = new ArrayList<>(expected); act = new ArrayList<>(act); Comparator<MultiDataSet> comp = new Comparator<MultiDataSet>() { @Override public int compare(MultiDataSet d1, MultiDataSet d2) { return Double.compare(d1.getFeatures(0).getDouble(0), d2.getFeatures(0).getDouble(0)); } }; Collections.sort(expected, comp); Collections.sort(act, comp); assertEquals(expected, act); }
public static JavaRDD<MultiDataSet> mapRRMDSI(JavaRDD<List<Writable>> rdd, RecordReaderMultiDataSetIterator iterator){ checkIterator(iterator, 1, 0); return mapRRMDSIRecords(rdd.map(new Function<List<Writable>,DataVecRecords>(){ @Override public DataVecRecords call(List<Writable> v1) throws Exception { return new DataVecRecords(Collections.singletonList(v1), null); } }), iterator); }
IteratorUtils { public static JavaRDD<MultiDataSet> mapRRMDSI(JavaRDD<List<Writable>> rdd, RecordReaderMultiDataSetIterator iterator){ checkIterator(iterator, 1, 0); return mapRRMDSIRecords(rdd.map(new Function<List<Writable>,DataVecRecords>(){ @Override public DataVecRecords call(List<Writable> v1) throws Exception { return new DataVecRecords(Collections.singletonList(v1), null); } }), iterator); } }
IteratorUtils { public static JavaRDD<MultiDataSet> mapRRMDSI(JavaRDD<List<Writable>> rdd, RecordReaderMultiDataSetIterator iterator){ checkIterator(iterator, 1, 0); return mapRRMDSIRecords(rdd.map(new Function<List<Writable>,DataVecRecords>(){ @Override public DataVecRecords call(List<Writable> v1) throws Exception { return new DataVecRecords(Collections.singletonList(v1), null); } }), iterator); }  }
IteratorUtils { public static JavaRDD<MultiDataSet> mapRRMDSI(JavaRDD<List<Writable>> rdd, RecordReaderMultiDataSetIterator iterator){ checkIterator(iterator, 1, 0); return mapRRMDSIRecords(rdd.map(new Function<List<Writable>,DataVecRecords>(){ @Override public DataVecRecords call(List<Writable> v1) throws Exception { return new DataVecRecords(Collections.singletonList(v1), null); } }), iterator); }  static JavaRDD<MultiDataSet> mapRRMDSI(JavaRDD<List<Writable>> rdd, RecordReaderMultiDataSetIterator iterator); static JavaRDD<MultiDataSet> mapRRMDSISeq(JavaRDD<List<List<Writable>>> rdd, RecordReaderMultiDataSetIterator iterator); static JavaRDD<MultiDataSet> mapRRMDSI(List<JavaRDD<List<Writable>>> rdds, List<JavaRDD<List<List<Writable>>>> seqRdds,
                                                  int[] rddsKeyColumns, int[] seqRddsKeyColumns, boolean filterMissing,
                                                  RecordReaderMultiDataSetIterator iterator); static JavaRDD<MultiDataSet> mapRRMDSIRecords(JavaRDD<DataVecRecords> rdd, RecordReaderMultiDataSetIterator iterator); }
IteratorUtils { public static JavaRDD<MultiDataSet> mapRRMDSI(JavaRDD<List<Writable>> rdd, RecordReaderMultiDataSetIterator iterator){ checkIterator(iterator, 1, 0); return mapRRMDSIRecords(rdd.map(new Function<List<Writable>,DataVecRecords>(){ @Override public DataVecRecords call(List<Writable> v1) throws Exception { return new DataVecRecords(Collections.singletonList(v1), null); } }), iterator); }  static JavaRDD<MultiDataSet> mapRRMDSI(JavaRDD<List<Writable>> rdd, RecordReaderMultiDataSetIterator iterator); static JavaRDD<MultiDataSet> mapRRMDSISeq(JavaRDD<List<List<Writable>>> rdd, RecordReaderMultiDataSetIterator iterator); static JavaRDD<MultiDataSet> mapRRMDSI(List<JavaRDD<List<Writable>>> rdds, List<JavaRDD<List<List<Writable>>>> seqRdds,
                                                  int[] rddsKeyColumns, int[] seqRddsKeyColumns, boolean filterMissing,
                                                  RecordReaderMultiDataSetIterator iterator); static JavaRDD<MultiDataSet> mapRRMDSIRecords(JavaRDD<DataVecRecords> rdd, RecordReaderMultiDataSetIterator iterator);  }