标签：

Flink 的流数据 API 编程指南

Flink 的流数据处理程序是常规的程序 ，通过再流数据上，实现了各种转换 (比如 过滤, 更新中间状态, 定义窗口, 聚合)。流数据可以来之多种数据源 (比如, 消息队列, socket 流, 文件). 通过sink组件落地流计算的最终结果，比如可以把数据落地文件系统，标准输出流比如命令行界面， Flink 的程序可以运行在多种上下文环境 ，可以单独只是Flink api，也可以嵌入其他程序. execution可以运行在本地的 JVM里, 也可以 运行在多台机器的集群中。为了创建你的流数据处理程序,，我们建议您从程序骨架开始，然后逐步添加您自己的transformations。后面的章节，是一些附加的操作，和一些高级功能。

• 例子程序
• 连接Flink
• 程序骨架
• 流数据抽象
• 延迟计算
• 转换
  • 物理分区
  • 任务链和资源组
• 指定key
• 提交函数给Flink
• 数据类型
• 数据源
• Execution配置
• 数据Sinks组件
• 调试
  • 本地运行环境
  • 集合框架数据源
• 窗口
  • 时间窗口
  • 指定过key的流数据的窗口
  • 为指定过key的流数据的窗口
• Execution参数
  • 故障容错机制
  • 并行度
  • 延迟控制
• 带状态的流数据计算
  • Checkpointing本地变量
  • 使用Key/Value的状态接口
  • 有状态的源的函数
  • 在迭代的任务中checkpoint状态
• 迭代计算
• 连接器
  • Apache Kafka
  • Elasticsearch
  • Hadoop FileSystem
  • RabbitMQ
  • Twitter Streaming API
  • Docker containers for connectors
• 打包程序和分布式运行
• 并行执行
• 执行计划

例子程序

下面是一个可运行的完整的例子  ，带窗口的流数据wordcount程序, 数据源来自一个每5秒一次的socket. 你可以复制黏贴并本地运行.

 1 public class WindowWordCount {      
 2     public static void main(String[] args) throws Exception {
 3         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 4         DataStream<Tuple2<String, Integer>> dataStream = env.socketTextStream("localhost", 9999)
 5                 .flatMap(new Splitter())
 6                 .keyBy(0)
 7                 .timeWindow(Time.of(5, TimeUnit.SECONDS))
 8                 .sum(1);
 9         dataStream.print();
10         env.execute("Window WordCount");
11         }
12     public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
13         @Override
14         public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
15             for (String word: sentence.split(" ")) {
16                 out.collect(new Tuple2<String, Integer>(word, 1));
17                 }
18             } 
19         } 
20 }

为了运行这个程序, 启动netcat在terminal中，敲下面这段:

nc -lk 9999

随便敲几个字. 这几个字将作为例子程序的输入. 如果你在5秒内，按了重复的字符，他们的count将会超过1. (如果你敲得不够快，可以提高5秒这个设置?).

Back to top

连接Flink

为了可以写Flink的代码, 你需要导入对应的语言的DataStream 的依赖包到你的工程里。

最简单的做法是使用quickstart 脚本: either for Java or for Scala. 你可以从模板中创建一个空工程 (a Maven Archetype), 这个工程已经准备好了一切编程所需的一切了.通过敲下面的代码，你可使用archetype 来手工的创建一个工程:

mvn archetype:generate /     -DarchetypeGroupId=org.apache.flink/     -DarchetypeArtifactId=flink-quickstart-java /     -DarchetypeVersion=1.0-SNAPSHOT

mvn archetype:generate /     -DarchetypeGroupId=org.apache.flink/     -DarchetypeArtifactId=flink-quickstart-scala /     -DarchetypeVersion=1.0-SNAPSHOT

这个archetypes依赖，稳定版本或者当前的版本(-SNAPSHOT).

如果你想为一个存在的maven工程添加Flink，在你的pom文件里添加下面这段依赖。

<dependency>   <groupId>org.apache.flink</groupId>   <artifactId>flink-streaming-java</artifactId>   <version>1.0-SNAPSHOT</version> </dependency> <dependency>   <groupId>org.apache.flink</groupId>   <artifactId>flink-clients</artifactId>   <version>1.0-SNAPSHOT</version> </dependency>

<dependency>   <groupId>org.apache.flink</groupId>   <artifactId>flink-streaming-scala</artifactId>   <version>1.0-SNAPSHOT</version> </dependency> <dependency>   <groupId>org.apache.flink</groupId>   <artifactId>flink-clients</artifactId>   <version>1.0-SNAPSHOT</version> </dependency>

程序骨架

getExecutionEnvironment()  createLocalEnvironment() createLocalEnvironment(int parallelism) createLocalEnvironment(int parallelism, Configuration customConfiguration)  createRemoteEnvironment(String host, int port, String... jarFiles) createRemoteEnvironment(String host, int port, int parallelism, String... jarFiles)

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();  DataStream<String> lines = env.socketTextStream("localhost", 9999)

DataStream<String> input = ...;  DataStream<Integer> intValues = input.map(new MapFunction<String, Integer>() {     @Override     public Integer map(String value) {         return Integer.parseInt(value);     } });

writeAsText(String path, ...) writeAsCsv(String path, ...) writeToSocket(String hostname, int port, ...)  print()  addSink(...)

env.execute();

def getExecutionEnvironment  def createLocalEnvironment(parallelism: Int =  Runtime.getRuntime.availableProcessors())  def createRemoteEnvironment(host: String, port: Int, jarFiles: String*) def createRemoteEnvironment(host: String, port: Int, parallelism: Int, jarFiles: String*)

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment  DataStream<String> lines = env.socketTextStream("localhost", 9999)

val input: DataStream[String] = ...  val mapped = input.map { x => x.toInt }

writeAsText(path: String, ...) writeAsCsv(path: String, ...) writeToSocket(hostname: String, port: Int, ...)  print()  addSink(...)

env.execute()

Back to top

流数据抽象

流数据是一种相同类型的无限的数据集合。

Transformations会返回不同子类类型的of DataStream，并且转换后的还可以继续transformations，比如 keyBy(…) 方法会返回KeyedDataStream ，这个也是流数据， 通过一个 key在本地做分区的流数据。还可以进行窗口操作。

Back to top

延迟执行

Flink的所有流数据程序是延迟执的。当main函数执行后, 数据加载和转换不是立刻执行的，相反的，每一步操作会加入一个执行计划.。直到evn执行execute方法来启动程序，这个执行计划才会执行。不论是本地执行还是在集群上执行.

延迟执行让你可以构建复杂的程序，并且让flink执行起来，像是个完整的计划单元。

Transformations

数据transformation让流数据产生新的流数据，. 程序可以结合多个流数据来构建复杂的应用拓扑结构。

这章给出了所有可用的transformations的详细说明。

DataStream<Integer> dataStream = //... dataStream.map(new MapFunction<Integer, Integer>() {
     @Override
     public Integer map(Integer value) throws Exception {
         return 2 * value;
     }
 });

dataStream.flatMap(new FlatMapFunction<String, String>() {     @Override     public void flatMap(String value, Collector<String> out)         throws Exception {         for(String word: value.split(" ")){             out.collect(word);         }     } });

dataStream.filter(new FilterFunction<Integer>() {     @Override     public boolean filter(Integer value) throws Exception {         return value != 0;     } });

dataStream.keyBy("someKey") // Key by field "someKey" dataStream.keyBy(0) // Key by the first element of a Tuple

keyedStream.reduce(new ReduceFunction<Integer>() {     @Override     public Integer reduce(Integer value1, Integer value2)     throws Exception {         return value1 + value2;     } });

keyedStream.fold(0, new ReduceFunction<Integer>() {   @Override   public Integer fold(Integer accumulator, Integer value)   throws Exception {       return accumulator + value;   } });

keyedStream.sum(0); keyedStream.sum("key"); keyedStream.min(0); keyedStream.min("key"); keyedStream.max(0); keyedStream.max("key"); keyedStream.minBy(0); keyedStream.minBy("key"); keyedStream.maxBy(0); keyedStream.maxBy("key");

dataStream.keyBy(0).window(TumblingTimeWindows.of(5, TimeUnit.SECONDS)); // Last 5 seconds of data

dataStream.windowAll(TumblingTimeWindows.of(Time.of(5, TimeUnit.SECONDS))); // Last 5 seconds of data

windowedStream.apply (new WindowFunction<Tuple2<String,Integer>,Integer>, Tuple, Window>() {     public void apply (Tuple tuple,             Window window,             Iterable<Tuple2<String, Integer>> values,             Collector<Integer> out) throws Exception {         int sum = 0;         for (value t: values) {             sum += t.f1;         }         out.collect (new Integer(sum));     } };

windowedStream.reduce (new ReduceFunction<Tuple2<String,Integer>() {     public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {         return new Tuple2<String,Integer>(value1.f0, value1.f1 + value2.f1);     } };

windowedStream.fold (new Tuple2<String,Integer>("Sum of all", 0),  new FoldFunction<Tuple2<String,Integer>() {     public Tuple2<String, Integer> fold(Tuple2<String, Integer> acc, Tuple2<String, Integer> value) throws Exception {         return new Tuple2<String,Integer>(acc.f0, acc.f1 + value.f1);     } };

windowedStream.sum(0); windowedStream.sum("key"); windowedStream.min(0); windowedStream.min("key"); windowedStream.max(0); windowedStream.max("key"); windowedStream.minBy(0); windowedStream.minBy("key"); windowedStream.maxBy(0); windowedStream.maxBy("key");

dataStream.union(otherStream1, otherStream2, ...);

dataStream.join(otherStream)     .where(0).equalTo(1)     .window(TumblingTimeWindows.of(Time.of(3, TimeUnit.SECONDS)))     .apply (new JoinFunction () {...});

dataStream.coGroup(otherStream)     .where(0).equalTo(1)     .window(TumblingTimeWindows.of(Time.of(3, TimeUnit.SECONDS)))     .apply (new CoGroupFunction () {...});

DataStream<Integer> someStream = //... DataStream<String> otherStream = //...  ConnectedStreams<Integer, String> connectedStreams = someStream.connect(otherStream);

connectedStreams.map(new CoMapFunction<Integer, String, Boolean>() {     @Override     public Boolean map1(Integer value) {         return true;     }      @Override     public Boolean map2(String value) {         return false;     } }); connectedStreams.flatMap(new CoFlatMapFunction<Integer, String, String>() {     @Override    public void flatMap1(Integer value, Collector<String> out) {        out.collect(value.toString());    }     @Override    public void flatMap2(String value, Collector<String> out) {        for (String word: value.split(" ")) {          out.collect(word);        }    } });

SplitStream<Integer> split = someDataStream.split(new OutputSelector<Integer>() {     @Override     public Iterable<String> select(Integer value) {         List<String> output = new ArrayList<String>();         if (value % 2 == 0) {             output.add("even");         }         else {             output.add("odd");         }         return output;     } });

SplitStream<Integer> split; DataStream<Integer> even = split.select("even"); DataStream<Integer> odd = split.select("odd"); DataStream<Integer> all = split.select("even","odd");

IterativeStream<Long> iteration = initialStream.iterate(); DataStream<Long> iterationBody = iteration.map (/*do something*/); DataStream<Long> feedback = iterationBody.filter(new FilterFunction<Long>(){     @Override     public boolean filter(Integer value) throws Exception {         return value > 0;     } }); iteration.closeWith(feedback); DataStream<Long> output = iterationBody.filter(new FilterFunction<Long>(){     @Override     public boolean filter(Integer value) throws Exception {         return value <= 0;     } });

stream.assignTimestamps (new TimeStampExtractor() {...});

dataStream.map { x => x * 2 }

dataStream.flatMap { str => str.split(" ") }

dataStream.filter { _ != 0 }

dataStream.keyBy("someKey") // Key by field "someKey" dataStream.keyBy(0) // Key by the first element of a Tuple

keyedStream.reduce { _ + _ }

keyedStream.fold { 0, _ + _ }

keyedStream.sum(0) keyedStream.sum("key") keyedStream.min(0) keyedStream.min("key") keyedStream.max(0) keyedStream.max("key") keyedStream.minBy(0) keyedStream.minBy("key") keyedStream.maxBy(0) keyedStream.maxBy("key")

dataStream.keyBy(0).window(TumblingTimeWindows.of(5, TimeUnit.SECONDS)) // Last 5 seconds of data // Last 5 seconds of data

dataStream.windowAll(TumblingTimeWindows.of(Time.of(5, TimeUnit.SECONDS))) // Last 5 seconds of data

windowedStream.apply { applyFunction }

windowedStream.reduce { _ + _ }

windowedStream.fold { 0, _ + _ }

windowedStream.sum(0) windowedStream.sum("key") windowedStream.min(0) windowedStream.min("key") windowedStream.max(0) windowedStream.max("key") windowedStream.minBy(0) windowedStream.minBy("key") windowedStream.maxBy(0) windowedStream.maxBy("key")

dataStream.union(otherStream1, otherStream2, ...)

dataStream.join(otherStream)     .where(0).equalTo(1)     .window(TumblingTimeWindows.of(Time.of(3, TimeUnit.SECONDS)))     .apply { ... }

dataStream.coGroup(otherStream)     .where(0).equalTo(1)     .window(TumblingTimeWindows.of(Time.of(3, TimeUnit.SECONDS)))     .apply {}

someStream : DataStream[Int] = ... otherStream : DataStream[String] = ...  val connectedStreams = someStream.connect(otherStream)

connectedStreams.map(     (_ : Int) => true,     (_ : String) => false ) connectedStreams.flatMap(     (_ : Int) => true,     (_ : String) => false )

val split = someDataStream.split(   (num: Int) =>     (num % 2) match {       case 0 => List("even")       case 1 => List("odd")     } )

val even = split select "even" val odd = split select "odd" val all = split.select("even","odd")

initialStream. iterate {   iteration => {     val iterationBody = iteration.map {/*do something*/}     (iterationBody.filter(_ > 0), iterationBody.filter(_ <= 0))   } } IterativeStream<Long> iteration = initialStream.iterate(); DataStream<Long> iterationBody = iteration.map (/*do something*/); DataStream<Long> feedback = iterationBody.filter ( _ > 0); iteration.closeWith(feedback);

stream.assignTimestamps { timestampExtractor }

The following transformations are available on data streams of Tuples:

DataStream<Tuple3<Integer, Double, String>> in = // [...] DataStream<Tuple2<String, Integer>> out = in.project(2,0);

val in : DataStream[(Int,Double,String)] = // [...] val out = in.project(2,0)

物理分区

Flink 也通过以下Functions对transformation后的流进行底层的控制（如果需要的话）

dataStream.partitionByHash("someKey"); dataStream.partitionByHash(0);

dataStream.partitionCustom(new Partitioner(){...}, "someKey"); dataStream.partitionCustom(new Partitioner(){...}, 0);

dataStream.partitionRandom();

dataStream.rebalance();

dataStream.broadcast();

dataStream.partitionByHash("someKey") dataStream.partitionByHash(0)

dataStream.partitionCustom(partitioner, "someKey") dataStream.partitionCustom(partitioner, 0)

dataStream.partitionRandom()

dataStream.rebalance()

dataStream.broadcast()

Task chaining and resource groups

Task chain 和资源组

chain两个接下来的transformations 的意思是将他们放在同一个线程里面执行，这会工作得更好。默认情况下，flink尽可能地将操作进行chain（例如：接下来的2个map transformation）。API提供了对chain的细粒度控制：

如果你想在整个job中禁用chain，那你可以使用StreamExecutionEnvironment.disableOperatorChaining() 。接下来的functions会提供更细粒度的控制。注意，这些functions只能用在transformation之后的DataStream中。比如，你可以这样：someStream.map(...).startNewChain()，但是不能这样：someStream.startNewChain()。

一个资源组就是Flink里的一个slot，参考slots，根据需要，你可以在对每一个slot进行手动隔离操作

someStream.filter(...).map(...).startNewChain().map(...);

someStream.map(...).disableChaining();

someStream.filter(...).startNewResourceGroup();

someStream.map(...).isolateResources();

someStream.filter(...).map(...).startNewChain().map(...)

someStream.map(...).disableChaining()

someStream.filter(...).startNewResourceGroup()

someStream.map(...).isolateResources()

Back to top

Specifying Keys

keyBy transformation 需要指定的键已经在DataStream上定义。

使用方式如下

DataStream<...> input = // [...] DataStream<...> windowed = input 	.keyBy(/*define key here*/) 	.window(/*define window here*/);

Flink的数据模型不是基于键值对的。因此，你不需要将dataStream转换为键值类型。“键”是虚拟的：他们就像function一样定义在数据流之上用来指导分组操作

参见 the relevant section of the DataSet API documentation来了解怎样去指定键。仅仅只需要将DataSet替换为DataStream和groupBy withkeyBy.

Passing Functions to Flink

一些transformations以用户自定义的额functions为参数

See the relevant section of the DataSet API documentation.

Back to top

Data Types

Flink places some restrictions on the type of elements that are used in DataStreams and in results of transformations. The reason for this is that the system analyzes the types to determine efficient execution strategies.

See the relevant section of the DataSet API documentation.

Back to top

Data Sources

Back to top

Execution Configuration

StreamExecutionEnvironment提供ExecutionConfig来设置job的运行时配置

See the relevant section of the DataSet API documentation.

以下是对应API：

• enableTimestamps() / disableTimestamps(): 给从源头发出来的event加上一个时间戳. areTimestampsEnabled().返回现在的时间

• setAutoWatermarkInterval(long milliseconds): Set the interval for automatic watermark emission.设置Watermarks 发送的时间间隔。可以通过getAutoWatermarkInterval()来获的目前的间隔。

Back to top

Data Sinks

Back to top

Debugging

Before running a streaming program in a distributed cluster, it is a good idea to make sure that the implemented algorithm works as desired. Hence, implementing data analysis programs is usually an incremental process of checking results, debugging, and improving.

Flink provides features to significantly ease the development process of data analysis programs by supporting local debugging from within an IDE, injection of test data, and collection of result data. This section give some hints how to ease the development of Flink programs.

Local Execution Environment

A LocalStreamEnvironment starts a Flink system within the same JVM process it was created in. If you start the LocalEnvironement from an IDE, you can set breakpoints in your code and easily debug your program.

A LocalEnvironment is created and used as follows:

final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();  DataStream<String> lines = env.addSource(/* some source */); // build your program  env.execute();

val env = StreamExecutionEnvironment.createLocalEnvironment()  val lines = env.addSource(/* some source */) // build your program  env.execute()

Collection Data Sources

Flink provides special data sources which are backed by Java collections to ease testing. Once a program has been tested, the sources and sinks can be easily replaced by sources and sinks that read from / write to external systems.

Collection data sources can be used as follows:

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);

val env = StreamExecutionEnvironment.createLocalEnvironment()  // Create a DataStream from a list of elements val myInts = env.fromElements(1, 2, 3, 4, 5)  // Create a DataStream from any Collection val data: Seq[(String, Int)] = ... val myTuples = env.fromCollection(data)  // Create a DataStream from an Iterator val longIt: Iterator[Long] = ... val myLongs = env.fromCollection(longIt)

Note: Currently, the collection data source requires that data types and iterators implement Serializable. Furthermore, collection data sources can not be executed in parallel ( parallelism = 1).

Back to top

窗口

时间窗口

时间窗口，通常是指一定时间内的一组event的组合。  时间窗口函数定义了时间的类型，目前支持三种不同的时间窗口：

• Processing time: Processing time是指当transformation发生时候的机器的时间， Processing time 是最简单的时间类型，也是性能最高的。. 但是，在分布式和异步环境下，机器时间，往往不一致和有很多不确定性。

• Event time: Event time是指每个event发生的时间 。 这份时间一般是当消息进入flink前，消息本身自带的。或者从消息的某个字段中抽取出来. 当使用event time的情况下,乱序的消息可以被适当的处理。. 举个例子, 在12分的时间窗口里，当一个10分钟的event在12分钟的时候到达了，transformation也会正确的处理这些乱了序的event。. Event time 的处理方式提供了可预测的结果。 , 但会带来更多的延迟, 因为乱序的消息需要被缓存起来到内存里。

• Ingestion time: Ingestion（食入，摄取） time 是当event进入到flink的时间。.当event消息进入到flink被分配到的task所在的机器上的时间，作为分配给event的时间，. Ingestion time比   processing time更有确定性和可预测性, 比event time有更低的延迟。因为不依赖外部系统。因此， Ingestion time 提供了一种处于两者之间的解决方案。 Ingestion time 其实可以说是 event time的一种特殊情况，实际上，ingestion time 和eventtime在flink的底层中的处理方式是一样的。 

当使用 event time时, transformations需要避免无限的等待event到达，Watermarks 提供了一种控制event time的偏移时间的机制。Watermarks是由 sources发射出来的. 一个watermark 带有一个确定的时间戳（long），比如转换后是2015-12-03 14:17:30 ，则表示，不会再有比这个时间更早的时间的消息会到达。

你可以通过下面的方式，选择你需要的时间语义。

env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime); env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime) env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime) env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

默认情况是TimeCharacteristic.ProcessingTime, 写一个processing time的语义的程序是不需要，再做其他事情。

如果要写一个event time语义的程序 , 需要做下面4个步骤：

• 1：设置event time的语义 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

• 2：使用DataStream.assignTimestamps(...) 告诉flink，时间戳和event的关联。比如说，第几个字段是时间戳。

• 设置 让时间戳有效，enableTimestamps(), 还有watermark的发射间隔。(setAutoWatermarkInterval(long milliseconds)) inExecutionConfig.

举个例子, 假设   我们有一个tuple的数据流, 并且里面的第一个字段是时间戳 (产生这些消息的系统赋予的，非flink),  并且我们知道处理时间和这个时间的落差不会超过1秒。

DataStream<Tuple4<Long,Integer,Double,String>> stream = //... stream.assignTimestamps(new TimestampExtractor<Tuple4<Long,Integer,Double,String>>{     @Override     public long extractTimestamp(Tuple4<Long,Integer,Double,String> element, long currentTimestamp) {         return element.f0;     }      @Override     public long extractWatermark(Tuple4<Long,Integer,Double,String> element, long currentTimestamp) {         return element.f0 - 1000;     }      @Override     public long getCurrentWatermark() {         return Long.MIN_VALUE;     } });

val stream: DataStream[(Long,Int,Double,String)] = null; stream.assignTimestampts(new TimestampExtractor[(Long, Int, Double, String)] {   override def extractTimestamp(element: (Long, Int, Double, String), currentTimestamp: Long): Long = element._1    override def extractWatermark(element: (Long, Int, Double, String), currentTimestamp: Long): Long = element._1 - 1000    override def getCurrentWatermark: Long = Long.MinValue })

如果你确定，你的时间戳一定是升序的，按顺序到达，你可以使用 AscendingTimestampExtractor, 系统会自动的发射watermark：

DataStream<Tuple4<Long,Integer,Double,String>> stream = //... stream.assignTimestamps(new AscendingTimestampExtractor<Tuple4<Long,Integer,Double,String>>{     @Override     public long extractAscendingTimestamp(Tuple4<Long,Integer,Double,String> element, long currentTimestamp) {         return element.f0;     } });

stream.extractAscendingTimestamp(record => record._1)

使用 ingestion time 语义，你需要：

 1：env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime).

 你可以想象一下，这个设置，其实是是event time的简写。因为source根据当前的机器时间，flink注入和发射都是在flink做的，所以flink可以推断后面的那些参数，所以自动做的。

Windows on Keyed Data Streams

Flink提供了好多方法，为KeyedStream定义windowon . 每个window 包含了同样的key的元素 。

Basic Window Constructs

Flink提供具有灵活性的通用窗口和普通情况下能通用的一些数据。在你需要自定义时间窗口之前，先看看这些预定义的能不能使用吧。

keyedStream.timeWindow(Time.of(5, TimeUnit.SECONDS));

keyedStream.timeWindow(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS));

keyedStream.countWindow(1000);

keyedStream.countWindow(1000, 100)

keyedStream.timeWindow(Time.of(5, TimeUnit.SECONDS))

keyedStream.timeWindow(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS))

keyedStream.countWindow(1000)

keyedStream.countWindow(1000, 100)

高级Window构造方式

这个机制可以定义出功能更丰富的窗口，相反的需要写更多的代码。 举个例子，下面是一个自定义的窗口，每个window持有最新的5秒并且每1秒滑动一次。 但是，当100个元素被添加到window后，window的execution函数，会被跟踪（触发）。之后每一次execution执行都会被跟踪（触发）。 window会保留10个元素:

keyedStream     .window(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS))     .trigger(Count.of(100))     .evictor(Count.of(10));

keyedStream     .window(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS))     .trigger(Count.of(100))     .evictor(Count.of(10))

构造一个自定义窗口的一般方式是，

（1）指定一个WindowAssigner, 

（2）指定 一个触发器Trigger (optionally),

（3）指定一个逐出器Evictor (optionally).

WindowAssigner定义了如何组织一个窗口 （时间或者个数） 一个window 元素的逻辑组合，有一个begin value，和一个end value。相应的，有一个begin time和end time. 带有时间戳的元素 。

举个例子，滑动时间窗口分配器，定义了5秒为一个单位，每1秒滑动一次，假设，以毫秒为单位，时间从0毫秒开始，然后我们有6个窗口： [0,5000], [1000,6000], [2000,7000], [3000, 8000], [4000, 9000], and [5000, 10000]. 每个进来的元素，根据他们的时间戳，被分配到这6个窗口中，可能出现在多个窗口里，比如带有2000 时间戳的元素，会被分配到前三个窗口。Flink 运行，会绑定在对应的窗口分配器，可以覆盖更多的场景. 你可以自定义你的window类型，通过继承WindowAssigner类。

stream.window(GlobalWindows.create());

stream.window(TumblingTimeWindows.of(Time.of(1, TimeUnit.SECONDS)));

stream.window(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS)));

stream.window(GlobalWindows.create)

stream.window(TumblingTimeWindows.of(Time.of(1, TimeUnit.SECONDS)))

stream.window(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS)))

Trigger定义了，跟在每一个的window后面的函数（sum，count），什么时候evaluated (“fires”)。如果没有指定trigger，就使用默认的trigger. Flink 自带了一组trigger，如果默认的trigger都没法满足你的应用，可以通过实现Trigger接口实现自己的trigger. 注意，如果使用自定义trigger后，会覆盖默认的trigger.

windowedStream.trigger(ProcessingTimeTrigger.create());

windowedStream.trigger(EventTimeTrigger.create());

windowedStream.trigger(ContinuousProcessingTimeTrigger.of(Time.of(5, TimeUnit.SECONDS)));

windowedStream.trigger(ContinuousEventTimeTrigger.of(Time.of(5, TimeUnit.SECONDS)));

windowedStream.trigger(CountTrigger.of(1000));

windowedStream.trigger(PurgingTrigger.of(CountTrigger.of(1000)));

windowedStream.trigger(new DeltaTrigger.of(5000.0, new DeltaFunction<Double>() {     @Override     public double getDelta (Double old, Double new) {         return (new - old > 0.01);     } }));

windowedStream.trigger(ProcessingTimeTrigger.create);

windowedStream.trigger(EventTimeTrigger.create);

windowedStream.trigger(ContinuousProcessingTimeTrigger.of(Time.of(5, TimeUnit.SECONDS)));

windowedStream.trigger(ContinuousEventTimeTrigger.of(Time.of(5, TimeUnit.SECONDS)));

windowedStream.trigger(CountTrigger.of(1000));

windowedStream.trigger(PurgingTrigger.of(CountTrigger.of(1000)));

windowedStream.trigger(DeltaTrigger.of(5000.0, { (old,new) => new - old > 0.01 }))

当trigger进行了fire之后, 并且执行sum和count之前, 有一个可选的逐出器可以移除保留元素。. Flink 自带了一组evictors ，你还可以通过实现Evictor接口，实现自定义的逐出器。.

triggeredStream.evict(TimeEvictor.of(Time.of(1, TimeUnit.SECONDS)));

triggeredStream.evict(CountEvictor.of(1000));

triggeredStream.evict(DeltaEvictor.of(5000, new DeltaFunction<Double>() {   public double (Double oldValue, Double newValue) {       return newValue - oldValue;   } }));

triggeredStream.evict(TimeEvictor.of(Time.of(1, TimeUnit.SECONDS)));

triggeredStream.evict(CountEvictor.of(1000));

windowedStream.evict(DeltaEvictor.of(5000.0, { (old,new) => new - old > 0.01 }))

Recipes for Building Windows

window 分配器，trigger，evictor的机制都很强大。这些机制让你可以定义各种不同类型的window。Flink’s的基本window其实是在这三个机制之上包了一层的，.下面是一些通用的端口是如何通过这三种机制来构造的 。

stream.countWindow(1000)

stream.window(GlobalWindows.create())   .trigger(CountTrigger.of(1000)   .evict(CountEvictor.of(1000)))

stream.countWindow(1000, 100)

stream.window(GlobalWindows.create())   .trigger(CountTrigger.of(1000)   .evict(CountEvictor.of(100)))

stream.timeWindow(Time.of(5, TimeUnit.SECONDS))

stream.window(TumblingTimeWindows.of((Time.of(5, TimeUnit.SECONDS)))   .trigger(EventTimeTrigger.create())

stream.timeWindow(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS))

stream.window(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS)))   .trigger(EventTimeTrigger.create())

stream.timeWindow(Time.of(5, TimeUnit.SECONDS))

stream.window(TumblingTimeWindows.of((Time.of(5, TimeUnit.SECONDS)))   .trigger(ProcessingTimeTrigger.create())

stream.timeWindow(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS))

stream.window(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS)))   .trigger(ProcessingTimeTrigger.create())

Windows on Unkeyed Data Streams

You也可以对普通流（stream，之前都是keyedStream）定义窗口。通过调用 the windowAll 这个transformation.这个stream 包含了所有keyed的stream, but 在一个单独的task里evaluated  (在一个单独的计算节点上). 定义trigger和evictor的语法是一样的：

nonKeyedStream     .windowAll(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS))     .trigger(Count.of(100))     .evictor(Count.of(10));

nonKeyedStream     .windowAll(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS))     .trigger(Count.of(100))     .evictor(Count.of(10))

基本的window 定义也适用于普通的nokey的windows：

nonKeyedStream.timeWindowAll(Time.of(5, TimeUnit.SECONDS));

nonKeyedStream.timeWindowAll(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS));

nonKeyedStream.countWindowAll(1000)

nonKeyedStream.countWindowAll(1000, 100)

nonKeyedStream.timeWindowAll(Time.of(5, TimeUnit.SECONDS));

nonKeyedStream.timeWindowAll(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS));

nonKeyedStream.countWindowAll(1000)

nonKeyedStream.countWindowAll(1000, 100)

Back to top

Execution Parameters

容错

Flink有检查点机制来恢复出现问题的job，此机制需要对信息进行持久化来支持之后需要的时候进行再次访问。

检查点机制会保存数据源的处理进度和用户自定义的状态来提供一次性处理语义

可以在StreamExecutionEnvironment上调用enableCheckpointing(n)来启用检查点机制，n代表检查点时间片的秒数

Other parameters for checkpointing include:

检查点中的的其它参数：

• 重试的次数：setNumberOfExecutionRerties() 这个方法返回这个job在失败后已经重启的次数。检查点启用打但是这个值没有显示设置的情况下，job会无限次地重新启动。
• 一次处理语义或至少一次语义：可以通过enableCheckpointing(n)来任选一种模式。一次处理语义对大多数应用来说是实用的，至少处理一次语义对某些低延时的（持续几毫秒）的应用来说会更好。

 docs on streaming fault tolerance 详细介绍了Flink的容错机制

当源添加了快照机制后Flink可以确保将一次处理语义更新为用户自定义的状态。现在只支持源头是kafka（和一些内置的数据生成器）。The following table lists the state update guarantees of Flink coupled with the bundled sources:

To guarantee end-to-end exactly-once record delivery (in addition to exactly-once updates), the data sink needs to take part in the snapshotting mechanism. The following table lists the delivery guarantees (assuming exactly-once state updates) of Flink coupled with bundled sinks:

Parallelism

You can control the number of parallel instances created for each operator by calling the operator.setParallelism(int) method.

Controlling Latency

By default, elements are not transferred on the network one-by-one (which would cause unnecessary network traffic) but are buffered. The size of the buffers (which are actually transferred between machines) can be set in the Flink config files. While this method is good for optimizing throughput, it can cause latency issues when the incoming stream is not fast enough. To control throughput and latency, you can useenv.setBufferTimeout(timeoutMillis) on the execution environment (or on individual operators) to set a maximum wait time for the buffers to fill up. After this time, the buffers are sent automatically even if they are not full. The default value for this timeout is 100 ms.

Usage:

LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(); env.setBufferTimeout(timeoutMillis);  env.genereateSequence(1,10).map(new MyMapper()).setBufferTimeout(timeoutMillis);

LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment env.setBufferTimeout(timeoutMillis)  env.genereateSequence(1,10).map(myMap).setBufferTimeout(timeoutMillis)

To maximize throughput, set setBufferTimeout(-1) which will remove the timeout and buffers will only be flushed when they are full. To minimize latency, set the timeout to a value close to 0 (for example 5 or 10 ms). A buffer timeout of 0 should be avoided, because it can cause severe performance degradation.

Back to top

Working with State

All transformations in Flink may look like functions (in the functional processing terminology), but are in fact stateful operators. You can make everytransformation (map, filter, etc) stateful by declaring local variables or using Flink’s state interface. You can register any local variable as managedstate by implementing an interface. In this case, and also in the case of using Flink’s native state interface, Flink will automatically take consistent snapshots of your state periodically, and restore its value in the case of a failure.

The end effect is that updates to any form of state are the same under failure-free execution and execution under failures.

First, we look at how to make local variables consistent under failures, and then we look at Flink’s state interface.

By default state checkpoints will be stored in-memory at the JobManager. For proper persistence of large state, Flink supports storing the checkpoints on file systems (HDFS, S3, or any mounted POSIX file system), which can be configured in the flink-conf.yaml or viaStreamExecutionEnvironment.setStateBackend(…).

Checkpointing Local Variables

Local variables can be checkpointed by using the Checkpointed interface.

When the user-defined function implements the Checkpointed interface, the snapshotState(…) and restoreState(…) methods will be executed to draw and restore function state.

In addition to that, user functions can also implement the CheckpointNotifier interface to receive notifications on completed checkpoints via thenotifyCheckpointComplete(long checkpointId) method. Note that there is no guarantee for the user function to receive a notification if a failure happens between checkpoint completion and notification. The notifications should hence be treated in a way that notifications from later checkpoints can subsume missing notifications.

For example the same counting, reduce function shown for OperatorStates by using the Checkpointed interface instead:

public class CounterSum implements ReduceFunction<Long>, Checkpointed<Long> {      // persistent counterprivatelong counter =0;      @Override     public Long reduce(Long value1, Long value2){         counter++;return value1 + value2;}// regularly persists state during normal operation     @Override     public Serializable snapshotState(long checkpointId,long checkpointTimestamp){return counter;}// restores state on recovery from failure     @Override     publicvoidrestoreState(Long state){         counter = state;}}

Using the Key/Value State Interface

The state interface gives access to key/value states, which are a collection of key/value pairs. Because the state is partitioned by the keys (distributed accross workers), it can only be used on the KeyedStream, created via stream.keyBy(…) (which means also that it is usable in all types of functions on keyed windows).

The handle to the state can be obtained from the function’s RuntimeContext. The state handle will then give access to the value mapped under the key of the current record or window - each key consequently has its own value.

The following code sample shows how to use the key/value state inside a reduce function. When creating the state handle, one needs to supply a name for that state (a function can have multiple states of different types), the type of the state (used to create efficient serializers), and the default value (returned as a value for keys that do not yet have a value associated).

public class CounterSum implements RichReduceFunction<Long> {      /** The state handle */     private OperatorState<Long> counter;      @Override     public Long reduce(Long value1, Long value2) {         counter.update(counter.value() + 1);         return value1 + value2;     }      @Override     public void open(Configuration config) {         counter = getRuntimeContext().getKeyValueState("myCounter", Long.class, 0L);     } }

State updated by this is usually kept locally inside the flink process (unless one configures explicitly an external state backend). This means that lookups and updates are process local and this very fast.

The important implication of having the keys set implicitly is that it forces programs to group the stream by key (via the keyBy() function), making the key partitioning transparent to Flink. That allows the system to efficiently restore and redistribute keys and state.

The Scala API has shortcuts that for stateful map() or flatMap() functions on KeyedStream, which give the state of the current key as an option directly into the function, and return the result with a state update:

val stream: DataStream[(String, Int)] = ...  val counts: DataStream[(String, Int)] = stream   .keyBy(_._1)   .mapWithState((in: (String, Int), count: Option[Int]) =>     count match {       case Some(c) => ( (in._1, c), Some(c + in._2) )       case None => ( (in._1, 0), Some(in._2) )     })

Stateful Source Functions

Stateful sources require a bit more care as opposed to other operators. In order to make the updates to the state and output collection atomic (required for exactly-once semantics on failure/recovery), the user is required to get a lock from the source’s context.

Some operators might need the information when a checkpoint is fully acknowledged by Flink to communicate that with the outside world. In this case see the flink.streaming.api.checkpoint.CheckpointNotifier interface.

State Checkpoints in Iterative Jobs

Flink currently only provides processing guarantees for jobs without iterations. Enabling checkpointing on an iterative job causes an exception. In order to force checkpointing on an iterative program the user needs to set a special flag when enabling checkpointing:env.enableCheckpointing(interval, force = true).

Please note that records in flight in the loop edges (and the state changes associated with them) will be lost during failure.

Back to top

Iterations

IterativeStream<Integer> iteration = input.iterate();

DataStream<Integer> iterationBody = iteration.map(/* this is executed many times */);

iteration.closeWith(iterationBody.filter(/* one part of the stream */)); DataStream<Integer> output = iterationBody.filter(/* some other part of the stream */);

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);   } });  iteration.closeWith(stillGreaterThanZero);  DataStream<Long> lessThanZero = minusOne.filter(new FilterFunction<Long>() {   @Override   public boolean filter(Long value) throws Exception {     return (value <= 0);   } });

val iteratedStream = someDataStream.iterate(   iteration => {     val iterationBody = iteration.map(/* this is executed many times */)     (tail.filter(/* one part of the stream */), tail.filter(/* some other part of the stream */)) })

val someIntegers: DataStream[Long] = env.generateSequence(0, 1000)  val iteratedStream = someIntegers.iterate(   iteration => {     val minusOne = iteration.map( v => v - 1)     val stillGreaterThanZero = minusOne.filter (_ > 0)     val lessThanZero = minusOne.filter(_ <= 0)     (stillGreaterThanZero, lessThanZero)   } )

Back to top

Connectors

Connectors provide code for interfacing with various third-party systems.

Currently these systems are supported:

• Apache Kafka (sink/source)
• Elasticsearch (sink)
• Hadoop FileSystem (sink)
• RabbitMQ (sink/source)
• Twitter Streaming API (source)

To run an application using one of these connectors, additional third party components are usually required to be installed and launched, e.g. the servers for the message queues. Further instructions for these can be found in the corresponding subsections. Docker containers are also provided encapsulating these services to aid users getting started with connectors.

Apache Kafka

This connector provides access to event streams served by Apache Kafka.

Flink provides special Kafka Connectors for reading and writing data to Kafka topics. The Flink Kafka Consumer integrates with Flink’s checkpointing mechanisms to provide different processing guarantees (most importantly exactly-once guarantees).

For exactly-once processing Flink can not rely on the auto-commit capabilities of the Kafka consumers. The Kafka consumer might commit offsets to Kafka which have not been processed successfully.

Please pick a package (maven artifact id) and class name for your use-case and environment. For most users, the flink-connector-kafka-083package and the FlinkKafkaConsumer082 class are appropriate.

Then, import the connector in your maven project:

<dependency>   <groupId>org.apache.flink</groupId>   <artifactId>flink-connector-kafka</artifactId>   <version>1.0-SNAPSHOT</version> </dependency>

Note that the streaming connectors are currently not part of the binary distribution. See how to link with them for cluster execution here.

Installing Apache Kafka

• Follow the instructions from Kafka’s quickstart to download the code and launch a server (launching a Zookeeper and a Kafka server is required every time before starting the application).
• On 32 bit computers this problem may occur.
• If the Kafka and Zookeeper servers are running on a remote machine, then the advertised.host.name setting in theconfig/server.properties file must be set to the machine’s IP address.

Kafka Consumer

The standard FlinkKafkaConsumer082 is a Kafka consumer providing access to one topic.

The following parameters have to be provided for the FlinkKafkaConsumer082(...) constructor:

1. The topic name
2. A DeserializationSchema
3. Properties for the Kafka consumer. The following properties are required:
   • “bootstrap.servers” (comma separated list of Kafka brokers)
   • “zookeeper.connect” (comma separated list of Zookeeper servers)
   • “group.id” the id of the consumer group

Example:

Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "localhost:9092"); properties.setProperty("zookeeper.connect", "localhost:2181"); properties.setProperty("group.id", "test"); DataStream<String> stream = env 	.addSource(new FlinkKafkaConsumer082<>("topic", new SimpleStringSchema(), properties)) 	.print();

val properties = new Properties(); properties.setProperty("bootstrap.servers", "localhost:9092"); properties.setProperty("zookeeper.connect", "localhost:2181"); properties.setProperty("group.id", "test"); stream = env     .addSource(new KafkaSource[String]("topic", new SimpleStringSchema(), properties))     .print

Kafka Consumers and Fault Tolerance

As Kafka persists all the data, a fault tolerant Kafka consumer can be provided.

The FlinkKafkaConsumer082 can read a topic, and if the job fails for some reason, the source will continue on reading from where it left off after a restart. For example if there are 3 partitions in the topic with offsets 31, 122, 110 read at the time of job failure, then at the time of restart it will continue on reading from those offsets, no matter whether these partitions have new messages.

To use fault tolerant Kafka Consumers, checkpointing of the topology needs to be enabled at the execution environment:

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(5000);

Also note that Flink can only restart the topology if enough processing slots are available to restart the topology. So if the topology fails due to loss of a TaskManager, there must still be enough slots available afterwards. Flink on YARN supports automatic restart of lost YARN containers.

Kafka Sink

A class providing an interface for sending data to Kafka.

The following arguments have to be provided for the KafkaSink(…) constructor in order:

1. Broker address (in hostname:port format, can be a comma separated list)
2. The topic name
3. Serialization schema

Example:

stream.addSink(new KafkaSink<String>("localhost:9092", "test", new SimpleStringSchema()));

stream.addSink(new KafkaSink[String]("localhost:9092", "test", new SimpleStringSchema))

The user can also define custom Kafka producer configuration for the KafkaSink with the constructor:

public KafkaSink(String zookeeperAddress, String topicId, Properties producerConfig,       SerializationSchema<IN, byte[]> serializationSchema)

public KafkaSink(String zookeeperAddress, String topicId, Properties producerConfig,       SerializationSchema serializationSchema)

If this constructor is used, the user needs to make sure to set the broker(s) with the “metadata.broker.list” property. Also the serializer configuration should be left default, and the serialization should be set via SerializationSchema.

The Apache Kafka official documentation can be found here.

Back to top

Elasticsearch

This connector provides a Sink that can write to an Elasticsearch Index. To use this connector, add the following dependency to your project:

<dependency>   <groupId>org.apache.flink</groupId>   <artifactId>flink-connector-elasticsearch</artifactId>   <version>1.0-SNAPSHOT</version> </dependency>

Note that the streaming connectors are currently not part of the binary distribution. See here for information about how to package the program with the libraries for cluster execution.

Installing Elasticsearch

Instructions for setting up an Elasticsearch cluster can be found here. Make sure to set and remember a cluster name. This must be set when creating a Sink for writing to your cluster

Elasticsearch Sink

The connector provides a Sink that can send data to an Elasticsearch Index.

The sink can use two different methods for communicating with Elasticsearch:

1. An embedded Node
2. The TransportClient

See here for information about the differences between the two modes.

This code shows how to create a sink that uses an embedded Node for communication:

DataStream<String> input = ...;  Map<String, String> config = Maps.newHashMap(); // This instructs the sink to emit after every element, otherwise they would be buffered config.put("bulk.flush.max.actions", "1"); config.put("cluster.name", "my-cluster-name");  input.addSink(new ElasticsearchSink<>(config, new IndexRequestBuilder<String>() {     @Override     public IndexRequest createIndexRequest(String element, RuntimeContext ctx) {         Map<String, Object> json = new HashMap<>();         json.put("data", element);          return Requests.indexRequest()                 .index("my-index")                 .type("my-type")                 .source(json);     } }));

val input: DataStream[String] = ...  val config = new util.HashMap[String, String] config.put("bulk.flush.max.actions", "1") config.put("cluster.name", "my-cluster-name")  text.addSink(new ElasticsearchSink(config, new IndexRequestBuilder[String] {   override def createIndexRequest(element: String, ctx: RuntimeContext): IndexRequest = {     val json = new util.HashMap[String, AnyRef]     json.put("data", element)     println("SENDING: " + element)     Requests.indexRequest.index("my-index").`type`("my-type").source(json)   } }))

Note how a Map of Strings is used to configure the Sink. The configuration keys are documented in the Elasticsearch documentation here. Especially important is the cluster.name parameter that must correspond to the name of your cluster.

Internally, the sink uses a BulkProcessor to send index requests to the cluster. This will buffer elements before sending a request to the cluster. The behaviour of the BulkProcessor can be configured using these config keys: * bulk.flush.max.actions: Maximum amount of elements to buffer * bulk.flush.max.size.mb: Maximum amount of data (in megabytes) to buffer * bulk.flush.interval.ms: Interval at which to flush data regardless of the other two settings in milliseconds

This example code does the same, but with a TransportClient:

DataStream<String> input = ...;  Map<String, String> config = Maps.newHashMap(); // This instructs the sink to emit after every element, otherwise they would be buffered config.put("bulk.flush.max.actions", "1"); config.put("cluster.name", "my-cluster-name");  List<TransportAddress> transports = new ArrayList<String>(); transports.add(new InetSocketTransportAddress("node-1", 9300)); transports.add(new InetSocketTransportAddress("node-2", 9300));  input.addSink(new ElasticsearchSink<>(config, transports, new IndexRequestBuilder<String>() {     @Override     public IndexRequest createIndexRequest(String element, RuntimeContext ctx) {         Map<String, Object> json = new HashMap<>();         json.put("data", element);          return Requests.indexRequest()                 .index("my-index")                 .type("my-type")                 .source(json);     } }));

val input: DataStream[String] = ...  val config = new util.HashMap[String, String] config.put("bulk.flush.max.actions", "1") config.put("cluster.name", "my-cluster-name")  val transports = new ArrayList[String] transports.add(new InetSocketTransportAddress("node-1", 9300)) transports.add(new InetSocketTransportAddress("node-2", 9300))  text.addSink(new ElasticsearchSink(config, transports, new IndexRequestBuilder[String] {   override def createIndexRequest(element: String, ctx: RuntimeContext): IndexRequest = {     val json = new util.HashMap[String, AnyRef]     json.put("data", element)     println("SENDING: " + element)     Requests.indexRequest.index("my-index").`type`("my-type").source(json)   } }))

The difference is that we now need to provide a list of Elasticsearch Nodes to which the sink should connect using a TransportClient.

More about information about Elasticsearch can be found here.

Back to top

Hadoop FileSystem

This connector provides a Sink that writes rolling files to any filesystem supported by Hadoop FileSystem. To use this connector, add the following dependency to your project:

<dependency>   <groupId>org.apache.flink</groupId>   <artifactId>flink-connector-filesystem</artifactId>   <version>1.0-SNAPSHOT</version> </dependency>

Note that the streaming connectors are currently not part of the binary distribution. See here for information about how to package the program with the libraries for cluster execution.

Rolling File Sink

The rolling behaviour as well as the writing can be configured but we will get to that later. This is how you can create a default rolling sink:

DataStream<String> input = ...;  input.addSink(new RollingSink<String>("/base/path"));

val input: DataStream[String] = ...  input.addSink(new RollingSink("/base/path"))

The only required parameter is the base path where the rolling files (buckets) will be stored. The sink can be configured by specifying a custom bucketer, writer and batch size.

By default the rolling sink will use the pattern "yyyy-MM-dd--HH" to name the rolling buckets. This pattern is passed to SimpleDateFormat with the current system time to form a bucket path. A new bucket will be created whenever the bucket path changes. For example, if you have a pattern that contains minutes as the finest granularity you will get a new bucket every minute. Each bucket is itself a directory that contains several part files: Each parallel instance of the sink will create its own part file and when part files get too big the sink will also create a new part file next to the others. To specify a custom bucketer use setBucketer() on a RollingSink.

The default writer is StringWriter. This will call toString() on the incoming elements and write them to part files, separated by newline. To specify a custom writer use setWriter() on a RollingSink. If you want to write Hadoop SequenceFiles you can use the providedSequenceFileWriter which can also be configured to use compression.

The last configuration option is the batch size. This specifies when a part file should be closed and a new one started. (The default part file size is 384 MB).

Example:

DataStream<Tuple2<IntWritable,Text>> input = ...;  RollingSink sink = new RollingSink<String>("/base/path"); sink.setBucketer(new DateTimeBucketer("yyyy-MM-dd--HHmm")); sink.setWriter(new SequenceFileWriter<IntWritable, Text>()); sink.setBatchSize(1024 * 1024 * 400); // this is 400 MB,  input.addSink(sink);

val input: DataStream[Tuple2[IntWritable, Text]] = ...  val sink = new RollingSink[String]("/base/path") sink.setBucketer(new DateTimeBucketer("yyyy-MM-dd--HHmm")) sink.setWriter(new SequenceFileWriter[IntWritable, Text]()) sink.setBatchSize(1024 * 1024 * 400) // this is 400 MB,  input.addSink(sink)

This will create a sink that writes to bucket files that follow this schema:

/base/path/{date-time}/part-{parallel-task}-{count}

Where date-time is the string that we get from the date/time format, parallel-task is the index of the parallel sink instance and count is the running number of part files that where created because of the batch size.

For in-depth information, please refer to the JavaDoc for RollingSink.

Back to top

RabbitMQ

This connector provides access to data streams from RabbitMQ. To use this connector, add the following dependency to your project:

<dependency>   <groupId>org.apache.flink</groupId>   <artifactId>flink-connector-rabbitmq</artifactId>   <version>1.0-SNAPSHOT</version> </dependency>

Note that the streaming connectors are currently not part of the binary distribution. See linking with them for cluster execution here.

Installing RabbitMQ

Follow the instructions from the RabbitMQ download page. After the installation the server automatically starts, and the application connecting to RabbitMQ can be launched.

RabbitMQ Source

A class providing an interface for receiving data from RabbitMQ.

The followings have to be provided for the RMQSource(…) constructor in order:

1. The hostname
2. The queue name
3. Deserialization schema

Example:

DataStream<String> stream = env 	.addSource(new RMQSource<String>("localhost", "hello", new SimpleStringSchema())) 	.print

stream = env     .addSource(new RMQSource[String]("localhost", "hello", new SimpleStringSchema))     .print

RabbitMQ Sink

A class providing an interface for sending data to RabbitMQ.

The followings have to be provided for the RMQSink(…) constructor in order:

1. The hostname
2. The queue name
3. Serialization schema

Example:

stream.addSink(new RMQSink<String>("localhost", "hello", new StringToByteSerializer()));

stream.addSink(new RMQSink[String]("localhost", "hello", new StringToByteSerializer))

More about RabbitMQ can be found here.

Back to top

Twitter Streaming API

Twitter Streaming API provides opportunity to connect to the stream of tweets made available by Twitter. Flink Streaming comes with a built-inTwitterSource class for establishing a connection to this stream. To use this connector, add the following dependency to your project:

<dependency>   <groupId>org.apache.flink</groupId>   <artifactId>flink-connector-twitter</artifactId>   <version>1.0-SNAPSHOT</version> </dependency>

Note that the streaming connectors are currently not part of the binary distribution. See linking with them for cluster execution here.

Authentication

In order to connect to Twitter stream the user has to register their program and acquire the necessary information for the authentication. The process is described below.

Acquiring the authentication information

First of all, a Twitter account is needed. Sign up for free at twitter.com/signup or sign in at Twitter’s Application Management and register the application by clicking on the “Create New App” button. Fill out a form about your program and accept the Terms and Conditions. After selecting the application, the API key and API secret (called consumerKey and sonsumerSecret in TwitterSource respectively) is located on the “API Keys” tab. The necessary access token data (token and secret) can be acquired here. Remember to keep these pieces of information secret and do not push them to public repositories.

Accessing the authentication information

Create a properties file, and pass its path in the constructor of TwitterSource. The content of the file should be similar to this:

#properties file for my app secret=*** consumerSecret=*** token=***-*** consumerKey=***

Constructors

The TwitterSource class has two constructors.

1. public TwitterSource(String authPath, int numberOfTweets); to emit finite number of tweets
2. public TwitterSource(String authPath); for streaming

Both constructors expect a String authPath argument determining the location of the properties file containing the authentication information. In the first case, numberOfTweets determines how many tweet the source emits.

Usage

In contrast to other connectors, the TwitterSource depends on no additional services. For example the following code should run gracefully:

DataStream<String> streamSource = env.addSource(new TwitterSource("/PATH/TO/myFile.properties"));

streamSource = env.addSource(new TwitterSource("/PATH/TO/myFile.properties"))

The TwitterSource emits strings containing a JSON code. To retrieve information from the JSON code you can add a FlatMap or a Map function handling JSON code. For example, there is an implementation JSONParseFlatMap abstract class among the examples. JSONParseFlatMap is an extension of the FlatMapFunction and has a

String getField(String jsonText, String field);

getField(jsonText : String, field : String) : String

function which can be use to acquire the value of a given field.

There are two basic types of tweets. The usual tweets contain information such as date and time of creation, id, user, language and many more details. The other type is the delete information.

Example

TwitterLocal is an example how to use TwitterSource. It implements a language frequency counter program.

Back to top

Docker containers for connectors

A Docker container is provided with all the required configurations for test running the connectors of Apache Flink. The servers for the message queues will be running on the docker container while the example topology can be run on the user’s computer.

Installing Docker

The official Docker installation guide can be found here. After installing Docker an image can be pulled for each connector. Containers can be started from these images where all the required configurations are set.

Creating a jar with all the dependencies

For the easiest setup, create a jar with all the dependencies of the flink-streaming-connectors project.

cd /PATH/TO/GIT/flink/flink-staging/flink-streaming-connectors mvn assembly:assembly ~~~bash  This creates an assembly jar under *flink-streaming-connectors/target*.  #### RabbitMQ Pull the docker image:  ~~~bash sudo docker pull flinkstreaming/flink-connectors-rabbitmq

To run the container, type:

sudo docker run -p 127.0.0.1:5672:5672 -t -i flinkstreaming/flink-connectors-rabbitmq

Now a terminal has started running from the image with all the necessary configurations to test run the RabbitMQ connector. The -p flag binds the localhost’s and the Docker container’s ports so RabbitMQ can communicate with the application through these.

To start the RabbitMQ server:

sudo /etc/init.d/rabbitmq-server start

To launch the example on the host computer, execute:

java -cp /PATH/TO/JAR-WITH-DEPENDENCIES org.apache.flink.streaming.connectors.rabbitmq.RMQTopology \ > log.txt 2> errorlog.txt

There are two connectors in the example. One that sends messages to RabbitMQ, and one that receives messages from the same queue. In the logger messages, the arriving messages can be observed in the following format:

<DATE> INFO rabbitmq.RMQTopology: String: <one> arrived from RMQ <DATE> INFO rabbitmq.RMQTopology: String: <two> arrived from RMQ <DATE> INFO rabbitmq.RMQTopology: String: <three> arrived from RMQ <DATE> INFO rabbitmq.RMQTopology: String: <four> arrived from RMQ <DATE> INFO rabbitmq.RMQTopology: String: <five> arrived from RMQ 

Apache Kafka

Pull the image:

sudo docker pull flinkstreaming/flink-connectors-kafka

To run the container type:

sudo docker run -p 127.0.0.1:2181:2181 -p 127.0.0.1:9092:9092 -t -i \ flinkstreaming/flink-connectors-kafka

Now a terminal has started running from the image with all the necessary configurations to test run the Kafka connector. The -p flag binds the localhost’s and the Docker container’s ports so Kafka can communicate with the application through these. First start a zookeeper in the background:

/kafka_2.9.2-0.8.1.1/bin/zookeeper-server-start.sh /kafka_2.9.2-0.8.1.1/config/zookeeper.properties \ > zookeeperlog.txt &

Then start the kafka server in the background:

/kafka_2.9.2-0.8.1.1/bin/kafka-server-start.sh /kafka_2.9.2-0.8.1.1/config/server.properties \  > serverlog.txt 2> servererr.txt &

To launch the example on the host computer execute:

java -cp /PATH/TO/JAR-WITH-DEPENDENCIES org.apache.flink.streaming.connectors.kafka.KafkaTopology \ > log.txt 2> errorlog.txt

In the example there are two connectors. One that sends messages to Kafka, and one that receives messages from the same queue. In the logger messages, the arriving messages can be observed in the following format:

<DATE> INFO kafka.KafkaTopology: String: (0) arrived from Kafka <DATE> INFO kafka.KafkaTopology: String: (1) arrived from Kafka <DATE> INFO kafka.KafkaTopology: String: (2) arrived from Kafka <DATE> INFO kafka.KafkaTopology: String: (3) arrived from Kafka <DATE> INFO kafka.KafkaTopology: String: (4) arrived from Kafka <DATE> INFO kafka.KafkaTopology: String: (5) arrived from Kafka <DATE> INFO kafka.KafkaTopology: String: (6) arrived from Kafka <DATE> INFO kafka.KafkaTopology: String: (7) arrived from Kafka <DATE> INFO kafka.KafkaTopology: String: (8) arrived from Kafka <DATE> INFO kafka.KafkaTopology: String: (9) arrived from Kafka 

Back to top

Program Packaging & Distributed Execution

See the relevant section of the DataSet API documentation.

Back to top

Parallel Execution

See the relevant section of the DataSet API documentation.

Back to top

Execution Plans

See the relevant section of the DataSet API documentation.

Back to top

<译>Flink编程指南

标签：

原文地址：http://www.cnblogs.com/Chuck-wu/p/5017436.html