开发者学堂课程【开源 Flink 极客训练营：Stream Processing with Apache Flink】学习笔记，与课程紧密联系，让用户快速学习知识。
课程地址：https://developer.aliyun.com/learning/course/760/detail/13338
Stream Processing with Apache Flink
7、Flink 连接器
外部系统:文件系统、数据库、消息队列.….…
Source 是否支持监测并接入更新(Dynamic or static)
Sink 是否支持更新已有结果(Append-only or support update)
Flink 里有两类比较关键处理的点，A 需要去连接外部系统从外部系统把数据读取到 Flink 的处理集群当中。第二类比较关键是 Sink 节点，需要汇总处理完的结果，把结果写入到某一个外部系统里面，这里的外部系统可以是一个文件系统，或者是一个数据库等等，每一个Flink 里面计算逻辑数据的输出，可以不把最终的数据写出外部系统，因为 Flink 里面有 state 是状态的概念，在中间计算的结果实际上是可以通过 state 的方式暴露给外部系统，可以没有专门的 Sink ，每一个 Flink 的应用肯定都有一个 Source ，必须从某一个地方去把数据读进来才能进行后续的处理，关于 Source 和 Sink 两类连接器，通常要关注的点是对于 Source 而言是比较关心数据接入是否支持监测并且持续感知到更新数据，把相应的更新数据传输到系统中来。
Flink 里对于文件有相应的连接器，比如 CSV 文件， CSV 文件的连接器在定义的时候可以指定一个参数，是否对 CS 文件读取一次，相当于取得文件系统里文件的快照读到系统里面，还是持续监测某一个目录的变化情况，目录里新添加去进去 CS 文件，系统可以感受到变化，并且再把新添加的数据读到系统中来，两者是一个静态的数据源，还有动态数据源的概念，去了解或者学习不同数据源的时候，一定要关注底，直接决定了后续的处理是有限还是持续不断连续的处理。类似的 Sink 是要写出的外部系统是否支持去更稀有的结果，对于部分的系统比如 Kafka要把数据写到 Kafka 里，Kafka 数据的写入是一种追加，比如不能够去修改一个已经写入到系统里的某一条记录和它对应比如要写出到一个数据库，数据库肯定支持数据的更新，比如本来数据是一，第二次对数据更新之后，把它结果变成了十，就可以通过一些手段把原来的一删掉，再把十作为更新之后的结果写出。这两个特性决定 Flink 里连接器面向静态的数据，面向动态的数据，非常关键的一些特点。关于系统连接器更多的信息可以去参照官方文档中的模块里面的内容。截图是1.1版本之后的文档，在 DataSource 连接器1.1版本里经历了相当于重构，在 Flink 27 里面进行了一些修改。对于 Table、SQL、API 层面上的连接器比起在 stream 层面上的连接器要承担更多的任务，比如会涉及到一些scheme 的定义，还有连接器是否支持一些谓词，或者投影操作的下推等等，因为支持这些功能是可以提高数据处理整体的性能。
（1）运行展示的例子
/**
*5. Basic DataStream API
*/
public static void gatastrea () throws Exception {
StreamExecutionEnvironment e=StreamExecutionEnvironfen t.getExecutionEnvironment()
DatastreamIntegers source e.addSourcel
new FronElementsFunction(Types.INT.createSerializer(e.getC onfig())，data)，Types.INT);
DataStreamInteger:ds=source.mop(v->*2).keyBy(value>1).sum(e);
ds.addSink(new PrintSinkFunctione0):
System.out.printIn(e.getExecutionPlan()); e.execute();
}
首先定义一个 environment 然后添加一个数据源，加完数据源之后对它进行一系列的转换操作，转换操作里添加一个最终的数据写出的 PrintSinkFunction ，作用是把最终的结果，每到来一条结果都把它输出到控制台标准输出，为了获取 DAG 把整个执行的计划输出，是非常有用的，在进行一些 bug 操作的时候可以看一看最终生成的执行计划是什么样，可以调用这个方法输出一个 json 格式的数据，最终可以调用 execute的方法，把上面所有的逻辑一起去执行。
（2）运行
/**
* Lessan 2 Streaning Processing with Apache Flink
*
* gauthor xccui
*/
public class Training2{
private static List data = Arrays.asList(1, 2, 3,4,5,6,7,8,9, 10)
public static void main(stringll args) throws Exception {
//System.out.println(declarative()); //System.out.println(inperativel());
//System.out.printin(inperative30)); //System.out.println[imperative20));
dataStream();
//state();
//processingTinewindow();
}
结果：
/Library/Java/JavaVirtualMachines/jdk1.8.8_101.jdk/Contents/Hone/bin/java.…
//输出的执行计划
{"nodes": [("id-:1,"type":"Source: Custom Source","pact":"Data Source","contents":"Source:CustomSource","parallelism":1),{"d":2,"type":Hap","pact":"Operator" ,"contents":"
3>4
3>16
3>36
3>38
3>48
3>66
3>74
3>90
3>96
3>110
操作其实是 rolling song 不停更新的操作，每到来一个新的数据就可以把它加到已有的 sum 结果里，每到一个新数据计算的结果，相当于都会更新，所以就看到了一连串的结果，最终所有的数据都发送并且处理完毕之后得到110，种单击完成的结果是完全一样的。
二、状态和时间
状态和时间涉及到对于整个流式处理应用的一些比较精细的操作步骤，非常重要。
1、有状态计算
所有的计算是可以简单的分为无状态的计算和有状态的计算。对于无状态计算相对而言比较容易，假设有一个算子没有脑子，记不住之前干过的所有事情，这个算子的工作是每次给一些数据，然后对于数据应用预先定义好的计算逻辑，比如算子是加法算式，每进来的一组数据都把它全部的加起来，最后把结果输出，这一类在函数式编程的思想里是纯函数，纯函数计算结果只和输入数据有关，不会和之前的计算或者外部状态不会对它产生任何影响呢，这一类计算比较容易。Flink 里的有状态计算，拿捡树枝小游戏举例，从业内人员来分析，做的非常好的是它自己记录了非常多的状态，比如有好几天没有登录上线，再上去和 NPC 对话的时候会告诉你已经很久没有上线，会记住之前上线的时间作为一种状态记录下来，再生成对话的内容的时候会考虑到状态的影响。不像以前相对老一些的游戏里面的 NPC 对话都是一种无状态，比如每次说话，所告诉的内容是完全一样的，实现有状态的计算要做的点是把之前的状态记录下来，然后再把状态注入到新的计算过程当中。实现方式有两种：
第一种可以把状态在数据进入算子之前提取出来，然后把状态数据和输入数据合并在一起，把状态作为一类，输入和原来输入数据合并在一起，然后同时注入到算子中最后得到输出。这种方式是在spark的structure streaming 里去应用。如果知道structure streaming里聚合的实现是这样，会提前的把之前的聚合结果添加到新一轮数据中，同时直接发送给聚合算。有一个非常大的好处是可以重用无状态的算子，比如算子在看到输入数据的时候不知道已经悄悄的把之前的状态数据给混入到新一批数据中，在输入的数据中算子是可以保持无状态，从而用无状态的算子实现有状态的计算，这是第一类方法。
第二类方法是 Sink ，算子本身是有状态的算子，每一次到来新的数据之后，来做计算的时候相当于同时要考虑新输的数据和已有状态对于计算过程的影响，最终把结果输出。计算引擎也应该像游戏一样变得越来越智能，甚至可以自动学习数据中潜在的规律，不断的优化自己处理逻辑、计算逻辑不断的提高处理的性能，脱离了状态的记录是不能完成的。
