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. 你可以复制黏贴并本地运行.

public class WindowWordCount {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStream<Tuple2<String, Integer>> dataStream = env.socketTextStream("localhost", 9999).flatMap(new Splitter()).keyBy(0).timeWindow(Time.of(5, TimeUnit.SECONDS)).sum(1);dataStream.print();env.execute("Window WordCount");}public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {@Overridepublic void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {for (String word: sentence.split(" ")) {out.collect(new Tuple2<String, Integer>(word, 1));}}}}

object WindowWordCount {def main(args: Array[String]) {val env = StreamExecutionEnvironment.getExecutionEnvironmentval text = env.socketTextStream("localhost", 9999)val counts = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }.map { (_, 1) }.keyBy(0).timeWindow(Time.of(5, TimeUnit.SECONDS)).sum(1)counts.printenv.execute("Window Stream WordCount")}
}

为了运行这个程序, 启动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>() {@Overridepublic Integer map(String value) {return Integer.parseInt(value);}
});

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

env.execute();

def getExecutionEnvironmentdef 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.getExecutionEnvironmentDataStream<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>() {@Overridepublic Integer map(Integer value) throws Exception {return 2 * value;}
});

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

dataStream.filter(new FilterFunction<Integer>() {@Overridepublic 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>() {@Overridepublic Integer reduce(Integer value1, Integer value2)throws Exception {return value1 + value2;}
});

keyedStream.fold(0, new ReduceFunction<Integer>() {@Overridepublic 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>() {@Overridepublic Boolean map1(Integer value) {return true;}@Overridepublic Boolean map2(String value) {return false;}
});
connectedStreams.flatMap(new CoFlatMapFunction<Integer, String, String>() {@Overridepublic void flatMap1(Integer value, Collector<String> out) {out.collect(value.toString());}@Overridepublic void flatMap2(String value, Collector<String> out) {for (String word: value.split(" ")) {out.collect(word);}}
});

SplitStream<Integer> split = someDataStream.split(new OutputSelector<Integer>() {@Overridepublic 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>(){@Overridepublic boolean filter(Integer value) throws Exception {return value > 0;}
});
iteration.closeWith(feedback);
DataStream<Long> output = iterationBody.filter(new FilterFunction<Long>(){@Overridepublic 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

Chaining two subsequent transformations means co-locating them within the same thread for better performance. Flink by default chains operators if this is possible (e.g., two subsequent map transformations). The API gives fine-grained control over chaining if desired:

Use StreamExecutionEnvironment.disableOperatorChaining() if you want to disable chaining in the whole job. For more fine grained control, the following functions are available. Note that these functions can only be used right after a DataStream transformation as they refer to the previous transformation. For example, you can use someStream.map(...).startNewChain(), but you cannot use someStream.startNewChain().

A resource group is a slot in Flink, see slots. You can manually isolate operators in separate slots if desired.

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

The keyBy transformation requires that a key is defined on its argument DataStream.

A DataStream is keyed as

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

The data model of Flink is not based on key-value pairs. Therefore, you do not need to physically pack the data stream types into keys and values. Keys are “virtual”: they are defined as functions over the actual data to guide the grouping operator.

See the relevant section of the DataSet API documentation on how to specify keys. Just replace DataSet with DataStream, and groupBy withkeyBy.

Passing Functions to Flink

Some transformations take user-defined functions as arguments.

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

The StreamExecutionEnvironment also contains the ExecutionConfig which allows to set job specific configuration values for the runtime.

See the relevant section of the DataSet API documentation.

Parameters in the ExecutionConfig that pertain specifically to the DataStream API are:

• enableTimestamps() / disableTimestamps(): Attach a timestamp to each event emitted from a source. areTimestampsEnabled() returns the current value.

• setAutoWatermarkInterval(long milliseconds): Set the interval for automatic watermark emission. You can get the current value withlong 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 programenv.execute();

val env = StreamExecutionEnvironment.createLocalEnvironment()val lines = env.addSource(/* some source */)
// build your programenv.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>>{@Overridepublic long extractTimestamp(Tuple4<Long,Integer,Double,String> element, long currentTimestamp) {return element.f0;}@Overridepublic long extractWatermark(Tuple4<Long,Integer,Double,String> element, long currentTimestamp) {return element.f0 - 1000;}@Overridepublic 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._1override def extractWatermark(element: (Long, Int, Double, String), currentTimestamp: Long): Long = element._1 - 1000override def getCurrentWatermark: Long = Long.MinValue
})

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

DataStream<Tuple4<Long,Integer,Double,String>> stream = //...
stream.assignTimestamps(new AscendingTimestampExtractor<Tuple4<Long,Integer,Double,String>>{@Overridepublic 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 offers a general window mechanism that provides flexibility, as well as a number of pre-defined windows for common use cases. See first if your use case can be served by the pre-defined windows below before moving to defining your own windows.

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>() {@Overridepublic 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

Fault Tolerance

Flink has a checkpointing mechanism that recovers streaming jobs after failues. The checkpointing mechanism requires a persistent or durablesource that can be asked for prior records again (Apache Kafka is a good example of a durable source).

The checkpointing mechanism stores the progress in the source as well as the user-defined state (see Working with State) consistently to provideexactly once processing guarantees.

To enable checkpointing, call enableCheckpointing(n) on the StreamExecutionEnvironment, where n is the checkpoint interval in milliseconds.

Other parameters for checkpointing include:

• Number of retries: The setNumberOfExecutionRerties() method defines how many times the job is restarted after a failure. When checkpointing is activated, but this value is not explicitly set, the job is restarted infinitely often.
• exactly-once vs. at-least-once: You can optionally pass a mode to the enableCheckpointing(n) method to choose between the two guarantee levels. Exactly-once is preferrable for most applications. At-least-once may be relevant for certain super-low-latency (consistently few milliseconds) applications.

The docs on streaming fault tolerance describe in detail the technique behind Flink’s streaming fault tolerance mechanism.

Flink can guarantee exactly-once state updates to user-defined state only when the source participates in the snapshotting mechanism. This is currently guaranteed for the Kafka source (and internal number generators), but not for other sources. 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 counterprivate long counter = 0;@Overridepublic Long reduce(Long value1, Long value2) {counter++;return value1 + value2;}// regularly persists state during normal operation@Overridepublic Serializable snapshotState(long checkpointId, long checkpointTimestamp) {return counter;}// restores state on recovery from failure@Overridepublic void restoreState(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;@Overridepublic Long reduce(Long value1, Long value2) {counter.update(counter.value() + 1);return value1 + value2;}@Overridepublic 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.

public static class CounterSource implements RichParallelSourceFunction<Long>, Checkpointed<Long> {/**  current offset for exactly once semantics */private long offset;/** flag for job cancellation */private volatile boolean isRunning = true;@Overridepublic void run(SourceContext<Long> ctx) {final Object lock = ctx.getCheckpointLock();while (isRunning) {// output and state update are atomicsynchronized (lock) {ctx.collect(offset);offset += 1;}}}@Overridepublic void cancel() {isRunning = false;}@Overridepublic Long snapshotState(long checkpointId, long checkpointTimestamp) {return offset;}@Overridepublic void restoreState(Long state) {offset = state;}
}

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