@Test public void testEnableSlotSharing() { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<Integer> sourceDataStream = env.fromElements(1, 2, 3); DataStream<Integer> mapDataStream = sourceDataStream.map(x -> x + 1); final List<Transformation<?>> transformations = new ArrayList<>(); transformations.add(sourceDataStream.getTransformation()); transformations.add(mapDataStream.getTransformation()); StreamGraph streamGraph = new StreamGraphGenerator( transformations, env.getConfig(), env.getCheckpointConfig()) .generate(); Collection<StreamNode> streamNodes = streamGraph.getStreamNodes(); for (StreamNode streamNode : streamNodes) { assertEquals(StreamGraphGenerator.DEFAULT_SLOT_SHARING_GROUP, streamNode.getSlotSharingGroup()); } }
public StreamGraph generate() { streamGraph = new StreamGraph(executionConfig, checkpointConfig); streamGraph.setStateBackend(stateBackend); streamGraph.setChaining(chaining); streamGraph.setScheduleMode(scheduleMode); streamGraph.setUserArtifacts(userArtifacts); streamGraph.setTimeCharacteristic(timeCharacteristic); streamGraph.setJobName(jobName); streamGraph.setBlockingConnectionsBetweenChains(blockingConnectionsBetweenChains); alreadyTransformed = new HashMap<>(); for (Transformation<?> transformation: transformations) { transform(transformation); } final StreamGraph builtStreamGraph = streamGraph; alreadyTransformed.clear(); alreadyTransformed = null; streamGraph = null; return builtStreamGraph; }
StreamGraphGenerator { public StreamGraph generate() { streamGraph = new StreamGraph(executionConfig, checkpointConfig); streamGraph.setStateBackend(stateBackend); streamGraph.setChaining(chaining); streamGraph.setScheduleMode(scheduleMode); streamGraph.setUserArtifacts(userArtifacts); streamGraph.setTimeCharacteristic(timeCharacteristic); streamGraph.setJobName(jobName); streamGraph.setBlockingConnectionsBetweenChains(blockingConnectionsBetweenChains); alreadyTransformed = new HashMap<>(); for (Transformation<?> transformation: transformations) { transform(transformation); } final StreamGraph builtStreamGraph = streamGraph; alreadyTransformed.clear(); alreadyTransformed = null; streamGraph = null; return builtStreamGraph; } }
StreamGraphGenerator { public StreamGraph generate() { streamGraph = new StreamGraph(executionConfig, checkpointConfig); streamGraph.setStateBackend(stateBackend); streamGraph.setChaining(chaining); streamGraph.setScheduleMode(scheduleMode); streamGraph.setUserArtifacts(userArtifacts); streamGraph.setTimeCharacteristic(timeCharacteristic); streamGraph.setJobName(jobName); streamGraph.setBlockingConnectionsBetweenChains(blockingConnectionsBetweenChains); alreadyTransformed = new HashMap<>(); for (Transformation<?> transformation: transformations) { transform(transformation); } final StreamGraph builtStreamGraph = streamGraph; alreadyTransformed.clear(); alreadyTransformed = null; streamGraph = null; return builtStreamGraph; } StreamGraphGenerator(List<Transformation<?>> transformations, ExecutionConfig executionConfig, CheckpointConfig checkpointConfig); }
StreamGraphGenerator { public StreamGraph generate() { streamGraph = new StreamGraph(executionConfig, checkpointConfig); streamGraph.setStateBackend(stateBackend); streamGraph.setChaining(chaining); streamGraph.setScheduleMode(scheduleMode); streamGraph.setUserArtifacts(userArtifacts); streamGraph.setTimeCharacteristic(timeCharacteristic); streamGraph.setJobName(jobName); streamGraph.setBlockingConnectionsBetweenChains(blockingConnectionsBetweenChains); alreadyTransformed = new HashMap<>(); for (Transformation<?> transformation: transformations) { transform(transformation); } final StreamGraph builtStreamGraph = streamGraph; alreadyTransformed.clear(); alreadyTransformed = null; streamGraph = null; return builtStreamGraph; } StreamGraphGenerator(List<Transformation<?>> transformations, ExecutionConfig executionConfig, CheckpointConfig checkpointConfig); static int getNewIterationNodeId(); StreamGraphGenerator setStateBackend(StateBackend stateBackend); StreamGraphGenerator setChaining(boolean chaining); StreamGraphGenerator setSlotSharingEnabled(boolean isSlotSharingEnabled); StreamGraphGenerator setScheduleMode(ScheduleMode scheduleMode); StreamGraphGenerator setUserArtifacts(Collection<Tuple2<String, DistributedCache.DistributedCacheEntry>> userArtifacts); StreamGraphGenerator setTimeCharacteristic(TimeCharacteristic timeCharacteristic); StreamGraphGenerator setDefaultBufferTimeout(long defaultBufferTimeout); StreamGraphGenerator setJobName(String jobName); StreamGraphGenerator setBlockingConnectionsBetweenChains(boolean blockingConnectionsBetweenChains); StreamGraph generate(); }
StreamGraphGenerator { public StreamGraph generate() { streamGraph = new StreamGraph(executionConfig, checkpointConfig); streamGraph.setStateBackend(stateBackend); streamGraph.setChaining(chaining); streamGraph.setScheduleMode(scheduleMode); streamGraph.setUserArtifacts(userArtifacts); streamGraph.setTimeCharacteristic(timeCharacteristic); streamGraph.setJobName(jobName); streamGraph.setBlockingConnectionsBetweenChains(blockingConnectionsBetweenChains); alreadyTransformed = new HashMap<>(); for (Transformation<?> transformation: transformations) { transform(transformation); } final StreamGraph builtStreamGraph = streamGraph; alreadyTransformed.clear(); alreadyTransformed = null; streamGraph = null; return builtStreamGraph; } StreamGraphGenerator(List<Transformation<?>> transformations, ExecutionConfig executionConfig, CheckpointConfig checkpointConfig); static int getNewIterationNodeId(); StreamGraphGenerator setStateBackend(StateBackend stateBackend); StreamGraphGenerator setChaining(boolean chaining); StreamGraphGenerator setSlotSharingEnabled(boolean isSlotSharingEnabled); StreamGraphGenerator setScheduleMode(ScheduleMode scheduleMode); StreamGraphGenerator setUserArtifacts(Collection<Tuple2<String, DistributedCache.DistributedCacheEntry>> userArtifacts); StreamGraphGenerator setTimeCharacteristic(TimeCharacteristic timeCharacteristic); StreamGraphGenerator setDefaultBufferTimeout(long defaultBufferTimeout); StreamGraphGenerator setJobName(String jobName); StreamGraphGenerator setBlockingConnectionsBetweenChains(boolean blockingConnectionsBetweenChains); StreamGraph generate(); static final int DEFAULT_LOWER_BOUND_MAX_PARALLELISM; static final ScheduleMode DEFAULT_SCHEDULE_MODE; static final TimeCharacteristic DEFAULT_TIME_CHARACTERISTIC; static final String DEFAULT_JOB_NAME; static final long DEFAULT_NETWORK_BUFFER_TIMEOUT; static final String DEFAULT_SLOT_SHARING_GROUP; }