@Test public void testMapsideJoin_RightSideIsEmpty() throws IOException { MRPipeline pipeline = new MRPipeline(MapsideJoinTest.class); PTable<Integer, String> customerTable = readTable(pipeline, "customers.txt"); PTable<Integer, String> orderTable = readTable(pipeline, "orders.txt"); PTable<Integer, String> filteredOrderTable = orderTable.parallelDo(new NegativeFilter(), orderTable.getPTableType()); PTable<Integer, Pair<String, String>> joined = MapsideJoin.join(customerTable, filteredOrderTable); List<Pair<Integer, Pair<String, String>>> materializedJoin = Lists.newArrayList(joined .materialize()); assertTrue(materializedJoin.isEmpty()); }
public static <K, U, V> PTable<K, Pair<U, V>> join(PTable<K, U> left, PTable<K, V> right) { if (!(right.getPipeline() instanceof MRPipeline)) { throw new CrunchRuntimeException("Map-side join is only supported within a MapReduce context"); } MRPipeline pipeline = (MRPipeline) right.getPipeline(); pipeline.materialize(right); pipeline.run(); ReadableSourceTarget<Pair<K, V>> readableSourceTarget = pipeline .getMaterializeSourceTarget(right); if (!(readableSourceTarget instanceof SourcePathTargetImpl)) { throw new CrunchRuntimeException("Right-side contents can't be read from a path"); } @SuppressWarnings("unchecked") SourcePathTargetImpl<Pair<K, V>> sourcePathTarget = (SourcePathTargetImpl<Pair<K, V>>) readableSourceTarget; Path path = sourcePathTarget.getPath(); DistributedCache.addCacheFile(path.toUri(), pipeline.getConfiguration()); MapsideJoinDoFn<K, U, V> mapJoinDoFn = new MapsideJoinDoFn<K, U, V>(path.toString(), right.getPType()); PTypeFamily typeFamily = left.getTypeFamily(); return left.parallelDo( "mapjoin", mapJoinDoFn, typeFamily.tableOf(left.getKeyType(), typeFamily.pairs(left.getValueType(), right.getValueType()))); }
MapsideJoin { public static <K, U, V> PTable<K, Pair<U, V>> join(PTable<K, U> left, PTable<K, V> right) { if (!(right.getPipeline() instanceof MRPipeline)) { throw new CrunchRuntimeException("Map-side join is only supported within a MapReduce context"); } MRPipeline pipeline = (MRPipeline) right.getPipeline(); pipeline.materialize(right); pipeline.run(); ReadableSourceTarget<Pair<K, V>> readableSourceTarget = pipeline .getMaterializeSourceTarget(right); if (!(readableSourceTarget instanceof SourcePathTargetImpl)) { throw new CrunchRuntimeException("Right-side contents can't be read from a path"); } @SuppressWarnings("unchecked") SourcePathTargetImpl<Pair<K, V>> sourcePathTarget = (SourcePathTargetImpl<Pair<K, V>>) readableSourceTarget; Path path = sourcePathTarget.getPath(); DistributedCache.addCacheFile(path.toUri(), pipeline.getConfiguration()); MapsideJoinDoFn<K, U, V> mapJoinDoFn = new MapsideJoinDoFn<K, U, V>(path.toString(), right.getPType()); PTypeFamily typeFamily = left.getTypeFamily(); return left.parallelDo( "mapjoin", mapJoinDoFn, typeFamily.tableOf(left.getKeyType(), typeFamily.pairs(left.getValueType(), right.getValueType()))); } }
MapsideJoin { public static <K, U, V> PTable<K, Pair<U, V>> join(PTable<K, U> left, PTable<K, V> right) { if (!(right.getPipeline() instanceof MRPipeline)) { throw new CrunchRuntimeException("Map-side join is only supported within a MapReduce context"); } MRPipeline pipeline = (MRPipeline) right.getPipeline(); pipeline.materialize(right); pipeline.run(); ReadableSourceTarget<Pair<K, V>> readableSourceTarget = pipeline .getMaterializeSourceTarget(right); if (!(readableSourceTarget instanceof SourcePathTargetImpl)) { throw new CrunchRuntimeException("Right-side contents can't be read from a path"); } @SuppressWarnings("unchecked") SourcePathTargetImpl<Pair<K, V>> sourcePathTarget = (SourcePathTargetImpl<Pair<K, V>>) readableSourceTarget; Path path = sourcePathTarget.getPath(); DistributedCache.addCacheFile(path.toUri(), pipeline.getConfiguration()); MapsideJoinDoFn<K, U, V> mapJoinDoFn = new MapsideJoinDoFn<K, U, V>(path.toString(), right.getPType()); PTypeFamily typeFamily = left.getTypeFamily(); return left.parallelDo( "mapjoin", mapJoinDoFn, typeFamily.tableOf(left.getKeyType(), typeFamily.pairs(left.getValueType(), right.getValueType()))); }  }
MapsideJoin { public static <K, U, V> PTable<K, Pair<U, V>> join(PTable<K, U> left, PTable<K, V> right) { if (!(right.getPipeline() instanceof MRPipeline)) { throw new CrunchRuntimeException("Map-side join is only supported within a MapReduce context"); } MRPipeline pipeline = (MRPipeline) right.getPipeline(); pipeline.materialize(right); pipeline.run(); ReadableSourceTarget<Pair<K, V>> readableSourceTarget = pipeline .getMaterializeSourceTarget(right); if (!(readableSourceTarget instanceof SourcePathTargetImpl)) { throw new CrunchRuntimeException("Right-side contents can't be read from a path"); } @SuppressWarnings("unchecked") SourcePathTargetImpl<Pair<K, V>> sourcePathTarget = (SourcePathTargetImpl<Pair<K, V>>) readableSourceTarget; Path path = sourcePathTarget.getPath(); DistributedCache.addCacheFile(path.toUri(), pipeline.getConfiguration()); MapsideJoinDoFn<K, U, V> mapJoinDoFn = new MapsideJoinDoFn<K, U, V>(path.toString(), right.getPType()); PTypeFamily typeFamily = left.getTypeFamily(); return left.parallelDo( "mapjoin", mapJoinDoFn, typeFamily.tableOf(left.getKeyType(), typeFamily.pairs(left.getValueType(), right.getValueType()))); }  static PTable<K, Pair<U, V>> join(PTable<K, U> left, PTable<K, V> right); }
MapsideJoin { public static <K, U, V> PTable<K, Pair<U, V>> join(PTable<K, U> left, PTable<K, V> right) { if (!(right.getPipeline() instanceof MRPipeline)) { throw new CrunchRuntimeException("Map-side join is only supported within a MapReduce context"); } MRPipeline pipeline = (MRPipeline) right.getPipeline(); pipeline.materialize(right); pipeline.run(); ReadableSourceTarget<Pair<K, V>> readableSourceTarget = pipeline .getMaterializeSourceTarget(right); if (!(readableSourceTarget instanceof SourcePathTargetImpl)) { throw new CrunchRuntimeException("Right-side contents can't be read from a path"); } @SuppressWarnings("unchecked") SourcePathTargetImpl<Pair<K, V>> sourcePathTarget = (SourcePathTargetImpl<Pair<K, V>>) readableSourceTarget; Path path = sourcePathTarget.getPath(); DistributedCache.addCacheFile(path.toUri(), pipeline.getConfiguration()); MapsideJoinDoFn<K, U, V> mapJoinDoFn = new MapsideJoinDoFn<K, U, V>(path.toString(), right.getPType()); PTypeFamily typeFamily = left.getTypeFamily(); return left.parallelDo( "mapjoin", mapJoinDoFn, typeFamily.tableOf(left.getKeyType(), typeFamily.pairs(left.getValueType(), right.getValueType()))); }  static PTable<K, Pair<U, V>> join(PTable<K, U> left, PTable<K, V> right);  }