FlinkPipelineExecutionEnvironment { public void translate(Pipeline pipeline) { this.flinkBatchEnv = null; this.flinkStreamEnv = null; final boolean hasUnboundedOutput = PipelineTranslationModeOptimizer.hasUnboundedOutput(pipeline); if (hasUnboundedOutput) { LOG.info("Found unbounded PCollection. Switching to streaming execution."); options.setStreaming(true); } prepareFilesToStageForRemoteClusterExecution(options); FlinkPipelineTranslator translator; if (options.isStreaming()) { this.flinkStreamEnv = FlinkExecutionEnvironments.createStreamExecutionEnvironment( options, options.getFilesToStage()); if (hasUnboundedOutput && !flinkStreamEnv.getCheckpointConfig().isCheckpointingEnabled()) { LOG.warn( "UnboundedSources present which rely on checkpointing, but checkpointing is disabled."); } translator = new FlinkStreamingPipelineTranslator(flinkStreamEnv, options); } else { this.flinkBatchEnv = FlinkExecutionEnvironments.createBatchExecutionEnvironment( options, options.getFilesToStage()); translator = new FlinkBatchPipelineTranslator(flinkBatchEnv, options); } pipeline.replaceAll(FlinkTransformOverrides.getDefaultOverrides(options)); translator.translate(pipeline); } FlinkPipelineExecutionEnvironment(FlinkPipelineOptions options); void translate(Pipeline pipeline); JobExecutionResult executePipeline();  }
@Test public void shouldRecognizeAndTranslateStreamingPipeline() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(TestFlinkRunner.class); options.setFlinkMaster("[auto]"); FlinkPipelineExecutionEnvironment flinkEnv = new FlinkPipelineExecutionEnvironment(options); Pipeline pipeline = Pipeline.create(); pipeline .apply(GenerateSequence.from(0).withRate(1, Duration.standardSeconds(1))) .apply( ParDo.of( new DoFn<Long, String>() { @ProcessElement public void processElement(ProcessContext c) throws Exception { c.output(Long.toString(c.element())); } })) .apply(Window.into(FixedWindows.of(Duration.standardHours(1)))) .apply(TextIO.write().withNumShards(1).withWindowedWrites().to("/dummy/path")); flinkEnv.translate(pipeline); }
@Test public void shouldUseDefaultTempLocationIfNoneSet() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(TestFlinkRunner.class); options.setFlinkMaster("clusterAddress"); FlinkPipelineExecutionEnvironment flinkEnv = new FlinkPipelineExecutionEnvironment(options); Pipeline pipeline = Pipeline.create(options); flinkEnv.translate(pipeline); String defaultTmpDir = System.getProperty("java.io.tmpdir"); assertThat(options.getFilesToStage(), hasItem(startsWith(defaultTmpDir))); }
@Test public void shouldUseTransformOverrides() { boolean[] testParameters = {true, false}; for (boolean streaming : testParameters) { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setStreaming(streaming); options.setRunner(FlinkRunner.class); FlinkPipelineExecutionEnvironment flinkEnv = new FlinkPipelineExecutionEnvironment(options); Pipeline p = Mockito.spy(Pipeline.create(options)); flinkEnv.translate(p); ArgumentCaptor<ImmutableList> captor = ArgumentCaptor.forClass(ImmutableList.class); Mockito.verify(p).replaceAll(captor.capture()); ImmutableList<PTransformOverride> overridesList = captor.getValue(); assertThat(overridesList.isEmpty(), is(false)); assertThat( overridesList.size(), is(FlinkTransformOverrides.getDefaultOverrides(options).size())); } }
@Test public void shouldProvideParallelismToTransformOverrides() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setStreaming(true); options.setRunner(FlinkRunner.class); FlinkPipelineExecutionEnvironment flinkEnv = new FlinkPipelineExecutionEnvironment(options); Pipeline p = Pipeline.create(options); p.apply(Create.of("test")).apply(TextIO.write().to("/tmp")); p = Mockito.spy(p); flinkEnv.translate(p); ArgumentCaptor<ImmutableList> captor = ArgumentCaptor.forClass(ImmutableList.class); Mockito.verify(p).replaceAll(captor.capture()); ImmutableList<PTransformOverride> overridesList = captor.getValue(); assertThat( overridesList, hasItem( new BaseMatcher<PTransformOverride>() { @Override public void describeTo(Description description) {} @Override public boolean matches(Object actual) { if (actual instanceof PTransformOverride) { PTransformOverrideFactory overrideFactory = ((PTransformOverride) actual).getOverrideFactory(); if (overrideFactory instanceof FlinkStreamingPipelineTranslator.StreamingShardedWriteFactory) { FlinkStreamingPipelineTranslator.StreamingShardedWriteFactory factory = (FlinkStreamingPipelineTranslator.StreamingShardedWriteFactory) overrideFactory; return factory.options.getParallelism() > 0; } } return false; } })); }
@Test public void shouldUseStreamingTransformOverridesWithUnboundedSources() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); FlinkPipelineExecutionEnvironment flinkEnv = new FlinkPipelineExecutionEnvironment(options); Pipeline p = Mockito.spy(Pipeline.create(options)); p.apply(GenerateSequence.from(0)); flinkEnv.translate(p); ArgumentCaptor<ImmutableList> captor = ArgumentCaptor.forClass(ImmutableList.class); Mockito.verify(p).replaceAll(captor.capture()); ImmutableList<PTransformOverride> overridesList = captor.getValue(); assertThat( overridesList, hasItem( PTransformOverride.of( PTransformMatchers.urnEqualTo(PTransformTranslation.CREATE_VIEW_TRANSFORM_URN), CreateStreamingFlinkView.Factory.INSTANCE))); }
@Test public void testTranslationModeOverrideWithUnboundedSources() { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setStreaming(false); FlinkPipelineExecutionEnvironment flinkEnv = new FlinkPipelineExecutionEnvironment(options); Pipeline pipeline = Pipeline.create(options); pipeline.apply(GenerateSequence.from(0)); flinkEnv.translate(pipeline); assertThat(options.isStreaming(), Matchers.is(true)); }
@Test public void testTranslationModeNoOverrideWithoutUnboundedSources() { boolean[] testArgs = new boolean[] {true, false}; for (boolean streaming : testArgs) { FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class); options.setRunner(FlinkRunner.class); options.setStreaming(streaming); FlinkPipelineExecutionEnvironment flinkEnv = new FlinkPipelineExecutionEnvironment(options); Pipeline pipeline = Pipeline.create(options); pipeline.apply(GenerateSequence.from(0).to(10)); flinkEnv.translate(pipeline); assertThat(options.isStreaming(), Matchers.is(streaming)); } }