CrunchFlinkPipelineFactory implements Serializable { @SuppressWarnings("unchecked") public DataStream<T> create(DataStream<MRecord> in, EvaluationPipeline<T> pipeline, Class<T> clazz) { List<DataStream<T>> results = new ArrayList<>(); for (SubStream<T> subStream: pipeline.getSubStreams()) { SingleOutputStreamOperator<MRecord> sortedIn = in .filter(createFilter(subStream.getPredicate())) .keyBy(x -> 0) .filter(createChannelFilter(subStream)) .assignTimestampsAndWatermarks(new ValueEventAssigner(subStream.getSortWindowMs())) .keyBy(x -> 0) .process(new SortFunction()); if (subStream.getRecordHandlers() != null && !subStream.getRecordHandlers().isEmpty()) { sortedIn .process( RecordProcessFunction.builder() .withRecordHandlers( subStream.getRecordHandlers() ) .build() ) .name("handler called") .addSink(new SinkFunction<Void>() { @Override public void invoke(Void value, Context context) { } }); } DataStream<T> dataStream = sortedIn .keyBy(x -> 0) .map(new ValuesMergeFunction<MRecord>()) .keyBy(x -> 0) .process( EvaluationProcessFunction.<T>builder() .withEvaluationFunctions(subStream.getEvalFunctions()) .build() ) .returns(clazz) .forward(); results.add(dataStream); } return results.stream() .reduce(DataStream::union) .orElse(null); }  @SuppressWarnings("unchecked") // is manually checked DataStream<T> create(DataStream<MRecord> in, EvaluationPipeline<T> pipeline, Class<T> clazz);  }
@Test @SuppressWarnings("unchecked") public void create() throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); List<SubStream<GenericEvent>> subStreams = new ArrayList<>(); List<EvaluationFunction<GenericEvent>> evaluationFunctions = new ArrayList<>(); List<RecordHandler> recordHandlers = new ArrayList<>(); final AtomicInteger i = new AtomicInteger(0); evaluationFunctions.add( new LambdaEvaluationFunction( ctx -> ((EvaluationContext<GenericEvent>)ctx).collect( new GenericEvent( System.currentTimeMillis(), "test" + i.addAndGet(1), "testSource" ) ), () -> new HashSet<>(Collections.singletonList("test")) ) ); recordHandlers.add(new NoopRecordHandler()); recordHandlers.add(new NoopRecordHandler()); recordHandlers.add(new NoopRecordHandler()); subStreams.add( SubStream.<GenericEvent>builder() .withIdentifier("testSubStream1") .withSortWindow(100L) .withPredicate(values -> true) .withRecordHandlers(recordHandlers) .withEvaluationFunctions(evaluationFunctions) .build() ); subStreams.add( SubStream.<GenericEvent>builder() .withIdentifier("testSubStream2") .withSortWindow(100L) .withPredicate(values -> true) .withRecordHandlers(recordHandlers) .withEvaluationFunctions(evaluationFunctions) .build() ); EvaluationPipeline<GenericEvent> pipeline = EvaluationPipeline.<GenericEvent>builder() .withIdentifier("testPipe") .withSubStreams(subStreams) .build(); DataStream<MRecord> in = createDataStream(env); in.print(); DataStream<GenericEvent> outEventDataStream = new CrunchFlinkPipelineFactory<GenericEvent>() .create(in, pipeline, GenericEvent.class); GenericEventCollectSink collectSink = new GenericEventCollectSink(); outEventDataStream.addSink(collectSink); env.execute("testJob"); Assert.assertEquals(200, GenericEventCollectSink.values.size()); }
@Test @SuppressWarnings("unchecked") public void createWithStringResult() throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); List<SubStream<String>> subStreams = new ArrayList<>(); List<EvaluationFunction<String>> evaluationFunctions = new ArrayList<>(); List<RecordHandler> recordHandlers = new ArrayList<>(); final AtomicInteger i = new AtomicInteger(0); evaluationFunctions.add(new LambdaEvaluationFunction( ctx -> ((EvaluationContext<String>)ctx).collect("I am the result"), () -> new HashSet<>(Collections.singleton("test")) )); recordHandlers.add(new NoopRecordHandler()); recordHandlers.add(new NoopRecordHandler()); recordHandlers.add(new NoopRecordHandler()); subStreams.add( SubStream.<String>builder() .withIdentifier("testSubStream1") .withSortWindow(100L) .withPredicate(values -> true) .withRecordHandlers(recordHandlers) .withEvaluationFunctions(evaluationFunctions) .build() ); subStreams.add( SubStream.<String>builder() .withIdentifier("testSubStream2") .withSortWindow(100L) .withPredicate(values -> true) .withRecordHandlers(recordHandlers) .withEvaluationFunctions(evaluationFunctions) .build() ); EvaluationPipeline<String> pipeline = EvaluationPipeline.<String>builder() .withIdentifier("testPipe") .withSubStreams(subStreams) .build(); DataStream<MRecord> in = createDataStream(env); in.print(); DataStream<String> outEventDataStream = new CrunchFlinkPipelineFactory() .create(in, pipeline, String.class); StringCollectSink collectSink = new StringCollectSink(); outEventDataStream.addSink(collectSink); env.execute("testJob"); Assert.assertEquals(200, StringCollectSink.values.size()); }