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)
