‘壹’ 4.一文搞定：Flink与Kafka之间的精准一次性
在上一篇文章当中，也算是比较详细且通俗的聊了聊Flink是如何通过checkpoint机制来完成数据精准一次性的实现的。并且也在上一章的结尾表示，要在接下来聊一聊Flink与它的铁哥们Kafaka之间，是如何实现数据的精准一次性消费的。
本次的聊法，还是要通过以kafka（source）->Flink,Flink(source)->Kafka来分别展开讨论。
kafka是一个具有数据保存、数据回放能力的消息队列，说白了就是kafka中的每一个数据，都有一个专门的标记作为标识。而在Flink消费kafka传入的数据的时候，source任务就能够将这个偏移量以算子状态的角色进行保存，写入到设定好的检查点中。这样一旦发生故障，Flink中的FlinkKafkaProce连接器就i能够按照自己保存的偏移量，自己去Kafka中重新拉取数据，也正是通过这种方式，就能够确保Kafka到Flink之间的精准一次性。
在上一篇文章当中，已经表明了，如果想要让输出端能够进行精准一次性消费，就需要使用到幂等性或者是事务。而事务中的两阶段提交是所有方案里面最好的实现。
其实Flink到Kafak之间也是采用了这种方式，具体的可以看一下ctrl进到FlinkKafkaProce连接器内部去看一看：
这也就表明了，当数据通过Flink发送给sink端Kafka的时候，是经历了两个阶段的处理的。第一阶段就是Flink向Kafka中插入数据，进入预提交阶段。当JobManager发送的Ckeckpoint保存成功信号过来之后，才会提交事务进行正式的数据发送，也就是让原来不可用的数据可以被使用了。
这个实现过程到目前阶段就很清晰了，它的主体流程无非就是在开启检查点之后，由JobManager向各个阶段的处理逻辑发送有关于检查点的barrier。所有的计算任务接收到之后，就会根据自己当前的状态做一个检查点保存。而当这个barrier来到sink任务的时候，sink就会开启一个事务，然后通过这个事务向外预写数据。直到Jobmanager来告诉它这一次的检查点已经保存完成了，sink就会进行第二次提交，数据也就算是成功写出了。
1.必须要保证检查点被打开了，如果检查点没有打开，那么之前说的一切话都是空谈。因为Flink默认检查点是关着的。
2.在FlinkKafakProcer连接器的构造函数中要传入参数，这个参数就是用来保证状态一致性的。就是在构造函数的最后一个参数输入如下：
3.配置Kafka读取数据的隔离级别
在kafka中有个配置，这个配置用来管理Kafka读取数据的级别。而这个配置默认是能够读取预提交阶段的数据的，所以如果你没改这个配置，那两阶段提交的第一阶段就是白费了。所以需要改一下这个配置，来更换一下隔离级别：
4.事务超时时间
这个配置也很有意思，大家试想一下。如果要进行两阶段提交，就要保证sink端支持事务，Kafka是支持事务的，但是像这个组件对于很多机制都有一个超时时间的概念，也就是说如果时间到了这个界限还没完成工作，那就会默认这个工作失败。Kafka中由这个概念，Flink中同样由这个概念。但是flink默认超时时间是1小时，而Kafka默认是15分钟，这就有可能出现检查点保存东西的时间大于15分钟，假如说是16分钟保存完成然后给sink发送检查点保存陈功可以提交事务的信号，但是这个时候Kafka已经认为事务失败，把之前的数据都扔了。那数据不就是丢失了么。所以说Kafka的超时时间要大于Flink的超时时间才好。
截止到目前为止，基本上把有关于状态维护的一些东西都说完了，有状态后端、有检查点。还通过检查点完成可端到端的数据精准一次性消费。但是想到这我又感觉，如果有学习进度比我差一些的，万一没办法很好的理解怎么办。所以在下一篇文章当中我就聊聊Flink中的“状态”到底是个什么东西，都有什么类型，都怎么去用。
‘贰’ 基于Flink的实时计算平台的构建
一、系统架构
1. 接入层
Canal、Flume、Kafka
针对业务系统数据，Canal监控Binlog日志，发送至kafka；
针对日志数据，由Flume来进行统一收集，并发送至kafka。
消息队列的数据既是离线数仓的原始数据，也是实时计算的原始数据，这样可以保证实时和离线的原始数据是统一的。
2. 计算层
Flink
有了源数据，在 计算层 经过Flink实时计算引擎做一些加工处理，然后落地到存储层中不同存储介质当中。
3. 存储层
HBase、Kafka、ES、Mysql、Hive、Redis
不同的 存储介质 是通过不同的应用场景来选择。
4. 数据应用层
风控、模型、图谱、大屏展示
通过存储层应用于不同的 数据应用 ，数据应用可能是我们的正式产品或者直接的业务系统
二、技术实现
1. 计算引擎
实时计算引擎的功能要求
提供高级 API，支持常见的数据操作比如关联聚合，最好是能支持 SQL
具有状态管理和自动支持久化方案，减少对存储的依赖
可靠的容错机制，低延时，最好能够保证Exactly-once
Flink的优势
Flink的API、容错机制与状态管理都满足实时数仓计算引擎的需求
Flink高吞吐、低延时的特性
端到端的Exactly-once
WaterMark&Event Time的支持
Flink 不仅支持了大量常用的 SQL 语句，还有丰富的数据类型、内置函数以及灵活的自定义函数，基本覆盖了我们的开发场景
2. 存储引擎
根据不同的业务场景，使用最适合的存储引擎：
Kafka主要用于中间数据表的存储
ES主要针对日志数据的存储和分析
HBase、Redis可用于维表存储
Hive用于数据校验
Mysql可以用于指标计算结果的存储
三、数据分层
数据源：目前数据源主要是Binlog，通过Canal监控各个业务系统的Mysql，将binlog发送至kafka。
ODS层：主要将Binlog数据存储至Kafka，这一层不对数据进行任何操作，存储最原始的数据，Binlog 日志在这一层为库级别，即：一个库的变更数据存放在同一个 Kafka Topic 中。
DWD层：主要对数据进行简单的清洗。拆分主题，将库级别的主题拆分为表级别；打平数据，将data数组格式打平。
DWS层：主要根据不同的业务的需求，将该需求所涉及到的表进行join所得。
APP层：根据指标计算需求，对数据进行处理后，存储HBase，为了方便模型查询，主要将表存储为索引表和明细表，直接对数据进行指标计算后，将计算结果存储到HBase。
四、数据监控及校验
1. 数据监控
目前数据的监控的架构是pushgateway + Prometheus + Grafana
数据监控主要是接入Flink的Metric，通过Grafana对Flink系统指标及自定义指标进行图形化界面的展示，对关键指标进行监控报警
2. 数据校验
目前数据的监控的架构是Grafana + Mysql
Grafana用于监控指标的展示及相关阈值数据的报警，Mysql主要用于监控数据的存储
将每个服务的source收到的数据、sink发出的数据，根据表的不同将数据关键字段写入mysql中，通过统计各个阶段各个表中的数据条数，对数据完整性进行监控校验，若出现数据缺时，先查找原因，然后指定时间戳重启服务
五、系统管理
元数据管理
表，字段元数据管理，实时感知元数据的变化，大幅度降低使用数据的成本。
系统配置
对应用启动参数及相关配置参数的管理，对任务进行灵活配置及管理。
血缘管理
主要是梳理实时计算平台中数据依赖关系，以及实时任务的依赖关系，从底层ODS到DWD再到DWS，以及APP层用到哪些数据，将整个链度串联起来。
六、问题及解决方案
1. 数据倾斜
由于要拆分主题，要以table为key对数据进行keyBy，但是由于每个表的数据量相差较大，会出现数据倾斜
解决方案：
加盐，给key加前缀
前缀不能随便加，为了保证同一id的数据在相同的分区中，所以根据id_table进行keyBy
2. 数据重复
任务在进行自动或手动重启时，为了保证数据不丢失，数据会出现重复计算的问题，如果下游只是对数据进行HBase存储的话，由于幂等性，这种重复可以解。但是，如果下游要对数据进行聚合，这样会导致数据被计算多次，影响计算结果的准确性
解决方案：
上游在对数据进行发送时，对kafka procer 进行 exactly once的设置
在对数据统计时进行数据去重
3. 数据延时
由于所处理的数据表的大小不一样，处理大表时，会出现数据延时的问题。
解决方案：
针对大表数据增加并行度
4.数据乱序
由于Flink kafka procer默认是根据hash对数据进行随机分区，kafka consumer在对数据进行消费时，每个分区消费速度不同，这样最终在存储数据时，就会出现乱序即相同的id会出现老数据覆盖新数据的问题
解决方案：
对kafka每个阶段进行自定义分区，将id相同的数据分到同一个分区，保证同一id的数据的有序性
由于整个数据处理过程中可能会出现shuffle，导数数据重新乱序，所以在对数据存储前对数据进行排序
对数据进行排序的关键点时要保证每条数据的唯一性，即要有标记数据先后顺序的字段
5 . 数据唯一标记（很重要）
由于要对数据进行去重或者排序，所以要保证数据的唯一性
解决办法：
使用时间戳不可以，因为数据量很大的情况下，同一时间会处理上百条数据
在最初发出数据的时候，为数据打上标记，使用 partition + offset + idx 的组合来确认数据的唯一性及顺序性
6. 数据可靠性
我们对服务重启或对服务升级时，可能会出现数据的丢失
解决方案：
结合Flink 的checkpoint及savepoint机制保证数据的可靠性
开启Flink的checkpoint机制，服务进行自动重启时，会自动读取上次保存在checkpoint中offset，或者我们指定offset进行数据消费
对服务进行升级时，先将服务的状态保存至savepoint中，重启时指定savepoint进行服务启动，保证数据不丢失
7. 无感升级
由于我们目前数据量比较庞大，且在对服务进行升级时，耗时较长，会影响调用方的使用。
解决办法：
在对服务进行升级时，将数据写入备用库，等数据追上且服务稳定运行后，再将存储库进行切换