devbox
笔触狂放9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

基于Canal与Flink实现数据实时增量同步(二)

作者：笔触狂放9 | 2024-06-15 18:27:43

赞

踩

flink 基于日志 同步mysql宽表数据到es

点击“蓝字”关注我吧

背景

在数据仓库建模中，未经任何加工处理的原始业务层数据，我们称之为ODS(Operational Data Store)数据。在互联网企业中，常见的ODS数据有业务日志数据（Log）和业务DB数据（DB）两类。对于业务DB数据来说，从MySQL等关系型数据库的业务数据进行采集，然后导入到Hive中，是进行数据仓库生产的重要环节。如何准确、高效地把MySQL数据同步到Hive中？一般常用的解决方案是批量取数并Load：直连MySQL去Select表中的数据，然后存到本地文件作为中间存储，最后把文件Load到Hive表中。这种方案的优点是实现简单，但是随着业务的发展，缺点也逐渐暴露出来：

  • 性能瓶颈：随着业务规模的增长，Select From MySQL -> Save to Localfile -> Load to Hive这种数据流花费的时间越来越长，无法满足下游数仓生产的时间要求。

  • 直接从MySQL中Select大量数据，对MySQL的影响非常大，容易造成慢查询，影响业务线上的正常服务。

  • 由于Hive本身的语法不支持更新、删除等SQL原语(高版本Hive支持，但是需要分桶+ORC存储格式)，对于MySQL中发生Update/Delete的数据无法很好地进行支持。

为了彻底解决这些问题，我们逐步转向CDC (Change Data Capture) + Merge的技术方案，即实时Binlog采集 + 离线处理Binlog还原业务数据这样一套解决方案。Binlog是MySQL的二进制日志，记录了MySQL中发生的所有数据变更，MySQL集群自身的主从同步就是基于Binlog做的。

实现思路

首先，采用Flink负责把Kafka上的Binlog数据拉取到HDFS上。

然后，对每张ODS表，首先需要一次性制作快照（Snapshot），把MySQL里的存量数据读取到Hive上，这一过程底层采用直连MySQL去Select数据的方式，可以使用Sqoop进行一次性全量导入。

最后，对每张ODS表，每天基于存量数据和当天增量产生的Binlog做Merge，从而还原出业务数据。

Binlog是流式产生的，通过对Binlog的实时采集，把部分数据处理需求由每天一次的批处理分摊到实时流上。无论从性能上还是对MySQL的访问压力上，都会有明显地改善。Binlog本身记录了数据变更的类型（Insert/Update/Delete），通过一些语义方面的处理，完全能够做到精准的数据还原。

实现方案

Flink处理Kafka的binlog日志

使用kafka source，对读取的数据进行JSON解析，将解析的字段拼接成字符串，符合Hive的schema格式，具体代码如下：

  1. package com.etl.kafka2hdfs;
  2.  
  3. import com.alibaba.fastjson.JSON;
  4. import com.alibaba.fastjson.JSONArray;
  5. import com.alibaba.fastjson.JSONObject;
  6. import com.alibaba.fastjson.parser.Feature;
  7. import org.apache.flink.api.common.functions.FilterFunction;
  8. import org.apache.flink.api.common.functions.MapFunction;
  9. import org.apache.flink.api.common.serialization.SimpleStringEncoder;
  10. import org.apache.flink.api.common.serialization.SimpleStringSchema;
  11. import org.apache.flink.core.fs.Path;
  12. import org.apache.flink.runtime.state.StateBackend;
  13. import org.apache.flink.runtime.state.filesystem.FsStateBackend;
  14. import org.apache.flink.streaming.api.datastream.DataStream;
  15. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  16. import org.apache.flink.streaming.api.environment.CheckpointConfig;
  17. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  18. import org.apache.flink.streaming.api.functions.sink.filesystem.RollingPolicy;
  19. import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
  20. import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
  21. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
  22.  
  23. import java.util.Map;
  24. import java.util.Properties;
  25.  
  26. /**
  27.  *  @Created with IntelliJ IDEA.
  28.  *  @author : jmx
  29.  *  @Date: 2020/3/27
  30.  *  @Time: 12:52
  31.  *  
  32.  */
  33. public class HdfsSink {
  34.     public static void main(String[] args) throws Exception {
  35.         String fieldDelimiter = ",";
  36.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  37.         env.setParallelism(1);
  38.  
  39.         // checkpoint
  40.         env.enableCheckpointing(10_000);
  41.         //env.setStateBackend((StateBackend) new FsStateBackend("file:///E://checkpoint"));
  42.         env.setStateBackend((StateBackend) new FsStateBackend("hdfs://kms-1:8020/checkpoint"));
  43.         CheckpointConfig config = env.getCheckpointConfig();
  44.         config.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
  45.  
  46.         // source
  47.         Properties props = new Properties();
  48.         props.setProperty("bootstrap.servers", "kms-2:9092,kms-3:9092,kms-4:9092");
  49.         // only required for Kafka 0.8
  50.         props.setProperty("zookeeper.connect", "kms-2:2181,kms-3:2181,kms-4:2181");
  51.         props.setProperty("group.id", "test123");
  52.         FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
  53.                 "qfbap_ods.code_city", new SimpleStringSchema(), props);
  54.         consumer.setStartFromEarliest();
  55.         DataStream<String> stream = env.addSource(consumer);
  56.  
  57.         // transform
  58.         SingleOutputStreamOperator<String> cityDS = stream
  59.                 .filter(new FilterFunction<String>() {
  60.                     // 过滤掉DDL操作
  61.                     @Override
  62.                     public boolean filter(String jsonVal) throws Exception {
  63.                         JSONObject record = JSON.parseObject(jsonVal, Feature.OrderedField);
  64.                         return record.getString("isDdl").equals("false");
  65.                     }
  66.                 })
  67.                 .map(new MapFunction<String, String>() {
  68.  
  69.                     @Override
  70.                     public String map(String value) throws Exception {
  71.                         StringBuilder fieldsBuilder = new StringBuilder();
  72.                         // 解析JSON数据
  73.                         JSONObject record = JSON.parseObject(value, Feature.OrderedField);
  74.  
  75.                         // 获取最新的字段值
  76.                         JSONArray data = record.getJSONArray("data");
  77.  
  78.                         // 遍历，字段值的JSON数组，只有一个元素
  79.                         for (int i = 0; i < data.size(); i++) {
  80.  
  81.                             // 获取到JSON数组的第i个元素
  82.                             JSONObject obj = data.getJSONObject(i);
  83.  
  84.                             if (obj != null) {
  85.  
  86.                                 fieldsBuilder.append(record.getLong("id")); // 序号id
  87.                                 fieldsBuilder.append(fieldDelimiter); // 字段分隔符
  88.                                 fieldsBuilder.append(record.getLong("es")); //业务时间戳
  89.                                 fieldsBuilder.append(fieldDelimiter);
  90.                                 fieldsBuilder.append(record.getLong("ts")); // 日志时间戳
  91.                                 fieldsBuilder.append(fieldDelimiter);
  92.                                 fieldsBuilder.append(record.getString("type")); // 操作类型
  93.                                 for (Map.Entry<String, Object> entry : obj.entrySet()) {
  94.  
  95.                                     fieldsBuilder.append(fieldDelimiter);
  96.                                     fieldsBuilder.append(entry.getValue()); // 表字段数据
  97.                                 }
  98.  
  99.                             }
  100.                         }
  101.                         return fieldsBuilder.toString();
  102.                     }
  103.  
  104.                 });
  105.  
  106.         //cityDS.print();
  107.         //stream.print();
  108.  
  109.         // sink
  110.         // 以下条件满足其中之一就会滚动生成新的文件
  111.         RollingPolicy<String, String> rollingPolicy = DefaultRollingPolicy.create()
  112.                 .withRolloverInterval(60L * 1000L) //滚动写入新文件的时间，默认60s。根据具体情况调节
  113.                 .withMaxPartSize(1024 * 1024 * 128L) //设置每个文件的最大大小 ,默认是128M，这里设置为128M
  114.                 .withInactivityInterval(60L * 1000L) //默认60秒,未写入数据处于不活跃状态超时会滚动新文件
  115.                 .build();
  116.  
  117.         StreamingFileSink<String> sink = StreamingFileSink
  118.                 //.forRowFormat(new Path("file:///E://binlog_db/city"), new SimpleStringEncoder<String>())
  119.                 .forRowFormat(new Path("hdfs://kms-1:8020/binlog_db/code_city_delta"), new SimpleStringEncoder<String>())
  120.                 .withBucketAssigner(new EventTimeBucketAssigner())
  121.                 .withRollingPolicy(rollingPolicy)
  122.                 .withBucketCheckInterval(1000)  // 桶检查间隔，这里设置1S
  123.                 .build();
  124.  
  125.         cityDS.addSink(sink);
  126.         env.execute();
  127.     }
  128. }

对于Flink Sink到HDFS，StreamingFileSink 替代了先前的 BucketingSink，用来将上游数据存储到 HDFS 的不同目录中。它的核心逻辑是分桶，默认的分桶方式是 DateTimeBucketAssigner，即按照处理时间分桶。处理时间指的是消息到达 Flink 程序的时间，这点并不符合我们的需求。因此，我们需要自己编写代码将事件时间从消息体中解析出来，按规则生成分桶的名称，具体代码如下：

  1. package com.etl.kafka2hdfs;
  2.  
  3. import org.apache.flink.core.io.SimpleVersionedSerializer;
  4. import org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner;
  5. import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.SimpleVersionedStringSerializer;
  6. import java.text.SimpleDateFormat;
  7. import java.util.Date;
  8.  
  9. /**
  10.  *  @Created with IntelliJ IDEA.
  11.  *  @author : jmx
  12.  *  @Date: 2020/3/27
  13.  *  @Time: 12:49
  14.  *  
  15.  */
  16.  
  17. public class EventTimeBucketAssigner implements BucketAssigner<String, String> {
  18.  
  19.  
  20.     @Override
  21.     public String getBucketId(String element, Context context) {
  22.         String partitionValue;
  23.         try {
  24.             partitionValue = getPartitionValue(element);
  25.         } catch (Exception e) {
  26.             partitionValue = "00000000";
  27.         }
  28.         return "dt=" + partitionValue;//分区目录名称
  29.     }
  30.  
  31.     @Override
  32.     public SimpleVersionedSerializer<String> getSerializer() {
  33.  
  34.         return SimpleVersionedStringSerializer.INSTANCE;
  35.     }
  36.  
  37.     private String getPartitionValue(String element) throws Exception {
  38.  
  39.         // 取出最后拼接字符串的es字段值，该值为业务时间
  40.         long eventTime = Long.parseLong(element.split(",")[1]);
  41.         Date eventDate = new Date(eventTime);
  42.         return new SimpleDateFormat("yyyyMMdd").format(eventDate);
  43.  
  44.     }
  45. }

离线还原MySQL数据

经过上述步骤，即可将Binlog日志记录写入到HDFS的对应的分区中，接下来就需要根据增量的数据和存量的数据还原最新的数据。Hive 表保存在 HDFS 上，该文件系统不支持修改，因此我们需要一些额外工作来写入数据变更。常用的方式包括：JOIN、Hive 事务、或改用 HBase、kudu。

如昨日的存量数据code_city,今日增量的数据为code_city_delta，可以通过 FULL OUTER JOIN，将存量和增量数据合并成一张最新的数据表，并作为明天的存量数据：

  1. INSERT OVERWRITE TABLE code_city
  2. SELECT 
  3.         COALESCE( t2.id, t1.id ) AS id,
  4.         COALESCE ( t2.city, t1.city ) AS city,
  5.         COALESCE ( t2.province, t1.province ) AS province,
  6.         COALESCE ( t2.event_time, t1.event_time ) AS event_time 
  7. FROM
  8.         code_city t1
  9.         FULL OUTER JOIN (
  10. SELECT
  11.         id,
  12.         city,
  13.         province,
  14.         event_time 
  15. FROM
  16.         (-- 取最后一条状态数据
  17. SELECT
  18.         id,
  19.         city,
  20.         province,
  21.         dml_type,
  22.         event_time,
  23.         row_number ( ) over ( PARTITION BY id ORDER BY event_time DESC ) AS rank 
  24. FROM
  25.         code_city_delta 
  26. WHERE
  27.         dt = '20200324' 
  28.         ) temp 
  29. WHERE
  30.         rank = 1 
  31.         ) t2 ON t1.id = t2.id;

小结

本文主要从Binlog流式采集和基于Binlog的ODS数据还原两方面，介绍了通过Flink实现实时的ETL，此外还可以将binlog日志写入kudu、HBase等支持事务操作的NoSQL中，这样就可以省去数据表还原的步骤。本文是《基于Canal与Flink实现数据实时增量同步》的第二篇，关于canal解析Binlog日志写入kafka的实现步骤，参见《基于Canal与Flink实现数据实时增量同步一》。

完

refrence：

[1]https://tech.meituan.com/2018/12/06/binlog-dw.html

戳原文，更有料！

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/笔触狂放9/article/detail/723330
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号