汇总:数栈技术分享：详解FlinkX中的断点续传和实时采集

　　汇总:数栈技术分享：详解FlinkX中的断点续传和实时采集

　　Datastack 是云原生的——一站式数据平台 PaaS。我们在 github 和 gitee 上有一个有趣的开源项目：FlinkX。FlinkX 是基于 Flink 的统一数据同步工具。它可以是采集静态数据，也可以是采集实时变化的数据，是一个全局的、异构的、批流式的集成数据同步引擎。如果喜欢，请给我们一个star！星星！星星！

　　github开源项目：

　　gitee开源项目：

　　袋鼠云独创的一站式数据中心PaaS-Data Stack，涵盖数据中心建设过程中所需的各种工具（包括数据开发平台、数据资产平台、数据科学平台、数据服务引擎等），完整覆盖离线计算和实时计算应用帮助企业大大缩短数据价值的提取过程，提高数据价值的提取能力。

　　目前已经基于FlinkX实现了数据栈-离线开发平台（BatchWorks）中的数据离线同步任务和数据栈-实时开发平台（StreamWorks）中的数据实时采集任务. 数据离线采集和实时采集的基本原理是一样的，主要区别在于源流是否有界，所以使用了Flink的Stream API来实现这两种数据的同步场景实现数据同步的批流统一。

　　一、功能介绍

　　1.断点续传

　　断点续传是指数据同步任务在运行过程中由于各种原因失败。不需要重新同步数据，只需要从上次失败的位置继续同步即可。类似于由于网络原因下载文件失败。无需重新下载文件，继续下载即可，可以大大节省时间和计算资源。断点续传是BatchWorks中数据同步任务的一个功能，需要结合任务的错误重试机制来完成。当任务运行失败时，它将在引擎中重试。重试时，会从上次失败时读取的位置继续读取数据，直到任务运行成功。

　　的

　　2.实时采集

　　实时采集是数据栈-实时开发平台（StreamWorks）中数据采集任务的一个功能。数据实时同步到目标数据源。除了实时数据变化之外，实时采集和离线数据同步的另一个区别是实时采集任务不会停止，任务会一直监控是否有数据源的变化。这与Flink任务一致，所以实时采集任务是数据栈流计算应用中的一种任务类型，配置过程与离线计算中的同步任务基本相同。

　　的

　　2. Flink中的Checkpoint机制

　　断点续传和实时采集都依赖于Flink的Checkpoint机制，我们先简单了解一下。

　　Checkpoint 是 Fl​​ink 容错机制的核心功能。它可以根据配置周期性的根据Stream中各个Operator的状态生成Snapshots，从而周期性的存储这些状态数据。当 Flink 程序意外崩溃时，可以重新运行。程序可以选择性地从这些Snapshots中恢复，从而纠正因故障导致的程序数据状态中断。

　　当Checkpoint被触发时，会在多个分布式Stream Source中插入一个Barrier标签，这些Barrier会随着Stream中的数据记录流向下游的Operator。当 Operator 收到 Barrier 时，它会暂停处理 Steam 中新收到的数据记录。因为一个Operator可能有多个input Streams，而每个Stream中都会有对应的Barrier，所以Operator会一直等到所有input Streams中的Barrier都到达。

　　当 Stream 中的所有 Barriers 都到达 Operator 时，所有的 Barriers 看起来都在同一时间点（说明它们已经对齐）。在等待所有Barrier到达的过程中，Operator的Buffer可能已经缓存了一些比Barrier更早到达Operator的数据记录（Outgoing Records），此时Operator会发出（Emit）这些数据记录(Outgoing Records)作为下游Operator的输入，最终发出（Emit）Barrier对应的Snapshot作为这个checkpoint的结果数据。

　　3.断点续传

　　一、先决条件

　　同步任务必须支持断点续传，对数据源有一些强制性要求：

　　1）数据源（这里特指关系型数据库）必须收录升序字段，如主键或日期类型字段。在同步过程中，会使用checkpoint机制记录该字段的值，任务恢复时会使用该字段构造查询条件。过滤已同步的数据。如果该字段的值不是升序排列，则任务恢复时筛选出的数据会出错，最终导致数据丢失或重复；

　　2）数据源必须支持数据过滤。如果不支持，任务无法从断点继续执行，会导致数据重复；

　　3）目标数据源必须支持事务，比如关系型数据库，文件类型的数据源也可以通过临时文件来支持。

　　2.任务操作的详细流程

　　我们用一个具体的任务来详细介绍整个过程。任务详情如下：

　　的

　　1）读取数据

　　读取数据时，首先要构造数据分片。构造数据分片就是根据通道索引和检查点记录的位置构造查询SQL。SQL模板如下：

　　select * from data_test

where id mod ${channel_num}=${channel_index}

and id > ${offset}

　　如果是第一次运行，或者上次任务失败时还没有触发checkpoint，那么offset不存在，具体查询sql可以根据offset和channel来判断：

　　当偏移量存在时

　　第一频道：

　　select * from data_test

where id mod 2=0

<p>

and id > ${offset_0};</p>

　　第二频道：

　　select * from data_test

where id mod 2=1

and id > ${offset_1};

　　当偏移量不存在时

　　第一频道：

　　select * from data_test

where id mod 2=0;

　　第二频道：

　　select * from data_test

where id mod 2=1;

　　数据分片构建完成后，各通道根据自己的数据分片读取数据。

　　2）写数据

　　在写数据之前，会先做几个操作：

　　一个。检查/data_test目录是否存在。如果该目录不存在，则创建该目录。如果目录存在，执行2个操作；

　　b. 判断是否覆盖写入数据，如果是，删除/data_test目录，再创建目录，如果不是，执行3次操作；

　　C。检查/data_test/.data目录是否存在。如果存在，先删除，再创建，保证没有其他任务异常故障留下的脏数据文件；

　　数据单条写入hdfs，不支持批量写入。数据会先写入/data_test/.data/目录，数据文件的命名格式为：

　　channelIndex.jobId.fileIndex

　　收录三部分：通道索引、jobId、文件索引。

　　3）当检查点被触发时

　　在 FlinkX 中，“state”表示标识字段 id 的值。我们假设触发checkpoint时两个channel的读写如图：

　　的

　　Checkpoint触发后，两个reader生成Snapshot记录读取状态，channel 0的状态为id=12，channel 1的状态为id=11。Snapshot生成后，在数据流中插入barrier，barrier跟随数据流向Writer。以 Writer_0 为例。Writer_0 接收Reader_0 和Reader_1 发送的数据。假设它先接收到 Reader_0 的屏障。此时Writer_0停止向HDFS写入数据，先将接收到的数据放入InputBuffer，等待Reader_1的barrier到来。然后把buffer里的数据全部写出来，然后生成Writer的快照。整个检查点结束后，记录的任务状态为：

　　读者_0：id=12

　　读者_1：id=11

　　Writer_0：id=无法确定

　　Writer_1：id=无法确定

　　任务状态会记录在配置的HDFS目录/flinkx/checkpoint/abc123中。因为每个Writer会收到两个Reader的数据，而且每个通道的数据读写速率可能不同，所以writer收到数据的顺序是不确定的，但这并不影响数据的准确性，因为读取数据构造查询sql只需要Reader记录的状态时，我们只需要保证数据真正写入HDFS即可。Writer在生成Snapshot之前，会进行一系列的操作，以确保将接收到的数据全部写入HDFS：

　　一个。关闭写入HDFS文件的数据流。这时候在/data_test/.data目录下会生成两个文件：

　　/data_test/.data/0.abc123.0

　　/data_test/.data/1.abc123.0

　　b. 将生成的两个数据文件移动到/data_test目录下；

　　C。更新文件名模板为：channelIndex.abc123.1；

　　快照生成后，任务继续读写数据。如果在生成快照的过程中出现异常，则任务会直接失败，从而不会生成本次快照，任务恢复时会从上次成功的快照开始恢复。

　　4）任务正常结束

　　当任务正常结束时，执行与生成快照时相同的操作，如关闭文件流、移动临时数据文件等。

　　

　　5）任务异常终止

　　如果任务异常结束，则假设任务结束时最后一条检查点记录的状态为：

　　读者_0：id=12 读者_1：id=11

　　那么当任务恢复时，会将每条通道记录的状态赋值到offset，再次读取数据时构造的SQL为：

　　第一频道：

　　select * from data_test

where id mod 2=0

and id > 12;

　　第二频道：

　　select * from data_test

where id mod 2=1

and id > 11;

　　这样就可以从上次失败的位置继续读取数据了。

　　3.支持断点续传的插件

　　理论上，只要支持过滤数据的数据源和支持事务的数据源都可以支持断点续传的功能，目前 FlinkX 支持的插件如下：

　　的

　　4. 实时 采集

　　目前 FlinkX 支持实时 采集 插件，包括 KafKa 和 binlog 插件。binlog插件是专门为mysql数据库的实时采集设计的。如果要支持其他数据源，只需要将数据发送到Kafka，然后再使用FlinkX的Kafka插件消费数据，比如oracle，只需要使用oracle的ogg发送数据到Kafka即可。这里具体讲解一下mysql的实时采集插件binlog。

　　1. 二进制日志

　　binlog是Mysql server层维护的二进制日志，与innodb引擎中的redo/undo log完全不同；主要用于记录更新或可能更新mysql数据的SQL语句，使用存储在磁盘上的“事务”。

　　binlog的主要功能有：

　　1）Replication：MySQL Replication在Master端开启binlog，Master将自己的二进制日志传递给slave并回放，达到主从数据一致性的目的；

　　2）数据恢复：通过mysqlbinlog工具恢复数据；

　　3）增量备份。

　　2.MySQL主备复制

　　仅仅有记录数据变化的binlog日志是不够的。我们还需要使用MySQL的主备复制功能：主备复制是指一台服务器作为主数据库服务器，另一台或多台服务器作为从数据库服务器。数据自动复制到从服务器。

　　的

　　主备复制的过程：

　　1）MySQL master将数据变化写入二进制日志（binary log，这里的记录称为binary log events，可以通过show binlog events查看）；

　　2）MySQL slave将master的二进制日志events复制到它的中继日志（relay log）；

　　3）MySQL slave重放relay log中的事件，将数据变化反映到自己的数据中。

　　3.写到蜂巢

　　binlog插件可以监控多个表的数据变化。解析后的数据收录表名信息。读取的数据可以写入目标数据库中的表，也可以根据数据中收录的表名信息写入。不同的表，目前只有Hive插件支持该功能。Hive插件目前只有write插件，功能是基于HDFS write插件，也就是说从binlog读取和写入hive也支持故障恢复的功能。

　　的

　　写入Hive的过程：

　　1）从数据中解析出MySQL表名，然后根据表名映射规则转换成对应的Hive表名；

　　2）检查Hive表是否存在，不存在则创建Hive表；

　　3）查询Hive表的相关信息，构造HdfsOutputFormat；

　　4）调用HdfsOutputFormat向HDFS写入数据。

　　汇总:数据采集自动数据采集软件

　　Free Data采集软件是一款绿色软件，可以根据关键词自动采集自动伪原创发布内容，无需编写复杂的采集规则。简单配置后即可实现24小时不间断采集、伪原创和发布。是站长维护网站的首选软件。内置全网发布接口 cms ，也可以txt格式直接导出到本地。是一款非常实用方便的采集软件。由于永久免费，得到了广大站长朋友的支持。是SEO圈子里的良心软件，给很多站长朋友带来了实实在在的流量和经济效益。

　　特点介绍：

　　1、自动去噪，可自动过滤标题内容中的图片、网址、电话、QQ、邮箱等信息；

　　2、本数据采集软件与传统采集模式的区别在于采集可以根据用户设置的关键词执行。采集的优点是可以根据采集关键词的不同搜索结果，自动采集最新发布的文章实现不执行采集在指定的一个或多个采集站点上，降低采集站点被搜索引擎判断为采集站点而被搜索引擎惩罚的风险。

　　

　　3、多种伪原创和优化方法，提高收录率和关键词排名标题插入、内容插入、自动内链、内容过滤、URL过滤、随机图片插入、定期发布，以及其他增强采集文章原创性能、提高搜索引擎收录、网站权重和关键词排名的方法。

　　一个搜索引擎，其核心价值是为用户提供他/她最需要的结果。搜索引擎对网民的需求有统计。对于网民需求很少或没有的内容，即使你是原创，搜索引擎也可能会忽略它，因为它不想将资源浪费在无意义的内容上。收录对于网友需求量大的内容应该有更多更快的内容，但是因为收录太多了，就算你是原创也可能很难挤进去排名。这么多用户选择使用采集！

　　1、使用数据采集软件需要注意网站结构规划吗？

　　1.网址设计。URL也可以收录关键词，比如你的网站是关于电脑的，你可以在你的URL中收录“PC”，因为在搜索引擎眼中它通常是“电脑”的同义词. URL不要太长，层级尽量不要超过4层，到此为止。

　　

　　2.栏目设计。栏目通常与导航相关联，设计时要考虑网站的整体主题，用户可能感兴趣的内容，栏目名称最好是网站的几个主要关键词 ，这也方便使用导航的权重。

　　3. 关键词布局。理论上，每一个内容页都应该将核心关键词、文章放在同一栏目下，并尽可能围绕关键词栏目展开。一个简单粗暴的方式就是直接使用关键词栏目中的长尾词。

　　2.根据数据量设置动态、伪静态和静态采集

　　这个不能一概而论，建议使用pseudo-static或者static。这三者的区别在于是否生成静态文件，URL格式是否动态。生成静态文件本质上是为了加快访问速度，减少数据库查询，但会不断增加占用空间；伪静态只是通过URL重写来修改URL，但实际上每次还是需要经过程序计算，查询数据库，然后输出页面。对于加速访问完全无效。动态和伪静态的区别仅在于URL，有问号和参数。

　　根据 网站 程序，数据库操作的效率可能会有所不同。一般来说，如果内容页数小于10000，页面打开速度比较快，数据量更大，达到50000、100000甚至更多，所以通常会考虑静态化。

　　看完这篇文章，如果您觉得不错，不妨采集或发送给需要的朋友和同事！你的一举一动都会成为小编源源不断的动力！