DataStream API 用来对data stream做转换，比如过滤，修改状态，定义窗口，聚合等。data source支持消息队列，socket流，文件等，data sink也支持文件，标准控制台输出等。

什么是DataStream

Flink中有一个类叫DataStream, 用来表示数据的集合，集合内数据不可更改，只能通过transformation生成新的DataStream. DataStream API的名字就来源于这个类。

Flink程序解析

Flink程序基本都包含如下五个部分：

获取执行环境
加载或创建初始数据
指定transformation操作
指定计算结果的输出
执行程序

执行环境

StreamExecutionEnvironment类提供了3个静态方法返回一个StreamExecutionEnvironment对象，默认只需要使用第一个，Flink会自行判断返回local或remote env

getExecutionEnvironment()

createLocalEnvironment()

createRemoteEnvironment(String host, int port, String... jarFiles)

Flink的程序是懒执行的，调用main方法后，数据加载和转换不会立即执行。而是先将每个操作添加到逻辑图中，并在env调用execute后才开始执行。

Data Sources

StreamExecutionEnvironment提供了几种预定义的source类型：

文件： readTextFile(path)，readFile(fileInputFormat, path)
socket：socketTextStream
集合：fromCollection(Collection)，fromElements(T ...)，generateSequence(from, to)
此外还可以使用addSource 方法自定义source类型，如从Kafka读取数据可以使用addSource(new FlinkKafkaConsumer<>(...))

Data Sinks

Flink内置的输出格式包括：

字符串：writeAsText()
excel: writeAsCsv(...)
文件：writeUsingOutputFormat()
socket: writeToSocket
或使用addSink 自定义
注意：write*()方法仅供调试使用，不参与checkpoint，因此只能保证至少一次而不是精确一次。数据落地到目标系统取决于OutputFormat的实现，因此输出到OutputFormat的数据可能不会立刻出现在目标系统里，异常发生时也有可能会丢失。
如果要保证到文件系统的精确一次的输出，建议使用StreamingFileSink。自定义的addSink的方法也可以通过checkpoint实现精确一次。

迭代流

IterativeStream能够实现事件在一系列转换操作中的迭代处理。即一个事件经过一系列转换后回到迭代流的头重走上述转换方法, 对迭代的数量没有限制，你可以指定哪些事件应该回到迭代头，哪些事件转发到下游输出流。

DataStream<Long> someIntegers = env.generateSequence(0, 1000);

// 创建一个迭代流
IterativeStream<Long> iteration = someIntegers.iterate();

// 定义对迭代流的转换操作
DataStream<Long> minusOne = iteration.map(new MapFunction<Long, Long>() {
  @Override
  public Long map(Long value) throws Exception {
    return value - 1 ;
  }
});

DataStream<Long> stillGreaterThanZero = minusOne.filter(new FilterFunction<Long>() {
  @Override
  public boolean filter(Long value) throws Exception {
    return (value > 0);
  }
});

// closeWith(feedbackStream)方法定义了迭代流的尾，入参feedbackStream会回到迭代头进行下一次迭代
iteration.closeWith(stillGreaterThanZero);

DataStream<Long> lessThanZero = minusOne.filter(new FilterFunction<Long>() {
  @Override
  public boolean filter(Long value) throws Exception {
    return (value <= 0);
  }
});

执行参数

StreamExecutionEnvironment 提供了一些接口来设置job运行时的参数，如setAutoWatermarkInterval指定水印的生成时间间隔，以及设置容错的参数和控制延迟的参数。元素在网络中是分组传输的，即缓存达到一定数目后才开始网络传输，如果buffer过大会导致高延迟，StreamExecutionEnvironment提供了setBufferTimeout(timeoutMillis)方法来指定buffer timeout时间，超时后即使buffer没满也会开始网络传输。

Debugging

在IDE中调试程序需要：

设置本地执行环境
用集合数据做数据源：集合数据源不能并行执行
采用Iterator Data Sink

final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();

// Create a DataStream from a list of elements
DataStream<Integer> myInts = env.fromElements(1, 2, 3, 4, 5);

// Create a DataStream from any Java collection
List<Tuple2<String, Integer>> data = ...
DataStream<Tuple2<String, Integer>> myTuples = env.fromCollection(data);

// Create a DataStream from an Iterator
Iterator<Long> longIt = ...
DataStream<Long> myLongs = env.fromCollection(longIt, Long.class);

DataStream<Tuple2<String, Integer>> myResult = ...
Iterator<Tuple2<String, Integer>> myOutput = DataStreamUtils.collect(myResult)

Overview