3.DataStream转换

DataStream：同一个类型的流元素，DataStream可以通过transformation转换成另外的DataStream,示例如下

@link DataStream#map

@link DataStream#filter

 StreamOperator:流式算子的基本接口，三个实现类

AbstractStreamOperator：

OneInputStreamOperator：

TwoInputStreamOperator：

/** * Basic interface for stream operators. Implementers would implement one of

* {@link org.apache.flink.streaming.api.operators.OneInputStreamOperator} or

* {@link org.apache.flink.streaming.api.operators.TwoInputStreamOperator} to create operators

* that process elements.

*

* <p>The class {@link org.apache.flink.streaming.api.operators.AbstractStreamOperator}

* offers default implementation for the lifecycle and properties methods.

*

* <p>Methods of {@code StreamOperator} are guaranteed not to be called concurrently. Also, if using

* the timer service, timer callbacks are also guaranteed not to be called concurrently with

* methods on {@code StreamOperator}.

*

* @param <OUT> The output type of the operator

*/

 4.DataStreamSink输出

/** * Adds the given sink to this DataStream. Only streams with sinks added

* will be executed once the {@link StreamExecutionEnvironment#execute()}

* method is called.

*

* @param sinkFunction

* The object containing the sink's invoke function.

* @return The closed DataStream.

*/publicDataStreamSink addSink(SinkFunction sinkFunction) {

// read the output type of the input Transform to coax out errors about MissingTypeInfo transformation.getOutputType();

// configure the type if neededif(sinkFunctioninstanceof InputTypeConfigurable) {

((InputTypeConfigurable) sinkFunction).setInputType(getType(), getExecutionConfig());

}

StreamSink sinkOperator =newStreamSink (clean(sinkFunction));

DataStreamSink sink =newDataStreamSink (this, sinkOperator);

getExecutionEnvironment().addOperator(sink.getTransformation());

return sink;

}

5.执行

/** * Executes the JobGraph of the on a mini cluster of ClusterUtil with a user

* specified name.

*

* @param jobName

* name of the job

* @return The result of the job execution, containing elapsed time and accumulators.

*/ @Override

publicJobExecutionResult execute(String jobName)throws Exception {

// transform the streaming program into a JobGraphStreamGraph streamGraph = getStreamGraph();

streamGraph.setJobName(jobName);

JobGraph jobGraph = streamGraph.getJobGraph();

jobGraph.setAllowQueuedScheduling(true);

Configuration configuration =new Configuration();

configuration.addAll(jobGraph.getJobConfiguration());

configuration.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "0");

// add (and override) the settings with what the user definedconfiguration.addAll(this.configuration);

if(!configuration.contains(RestOptions.BIND_PORT)) {

configuration.setString(RestOptions.BIND_PORT, "0");

}

intnumSlotsPerTaskManager = configuration.getInteger(TaskManagerOptions.NUM_TASK_SLOTS, jobGraph.getMaximumParallelism());

MiniClusterConfiguration cfg =new MiniClusterConfiguration.Builder()

.setConfiguration(configuration)

.setNumSlotsPerTaskManager(numSlotsPerTaskManager)

.build();

if (LOG.isInfoEnabled()) {

LOG.info("Running job on local embedded Flink mini cluster");

}

MiniCluster miniCluster =new MiniCluster(cfg);

try {

miniCluster.start();

configuration.setInteger(RestOptions.PORT, miniCluster.getRestAddress().get().getPort());

return miniCluster.executeJobBlocking(jobGraph);

}

finally {

transformations.clear();

miniCluster.close();

}

}

6.总结

　　Flink的执行方式类似于管道，它借鉴了数据库的一些执行原理，实现了自己独特的执行方式。

7.展望

Stream涉及的内容还包括Watermark，window等概念，因篇幅限制，这篇仅介绍flink DataStream API使用及原理。

下篇将介绍Watermark，下下篇是windows窗口计算。