一、数据同步方案

图片.png

二、技术选型

2.1 开源框架与阿里云架构对比

分类	阿里云架构	开源架构
数据采集	DTS、DataHub	canal、flume
数据传输	DTS、DataHub	kafka
数据存储	Rds、AnalyticDB	Mysql、HBase
数据计算	阿里云实时计算	Flink、spark

2.2 Binlog数据采集 Canal

• Canal项目地址：
  https://github.com/alibaba/canal

图片.png 图片.png

canal 工作原理

canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议
MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
canal 解析 binary log 对象(原始为 byte 流)

2.3 Binlog 实时传输 Kafka

图片.png

使用kafka 将canal 读取的binlog数据实时同步到Flink。由于kafka 可以设置多个Group，也可以设置多个partition，故消费时要进行处理，否则有可能出现同表数据进入不同Group，不同partition。故消费时要按照 database + table 将数据进行排序整理，保证同一个表放入同一个分区。

2.4 实时计算 Flink

图片.png

Flink的state管理和checkPoint 机制，可以保证Exactly-Once (同一数据经过多次操作还是同一条)

2.3.1 Flink 批处理和流处理

Flink CoreApis 通过一个底层引擎同时支持流处理和批处理.

图片.png

2.3.2 Flink 如何保证Exactly-Once

图片.png

Flink 可以保存每一个算子的state, 并且通过checkPoint机制对state进行保存。

2.3.3 Flink Table和SQL API

可以通过Flink Table API 和SQL API 进行编程。

• Table API
  接入数据源之后，可以直接将数据源转化为table, 并进行相应转换。
  //DataSet 转table, 指定字段名
  Table table = fbTableEnv.fromDataSet(ds2, "id,name");
  Table table02 = table.select("name");

/**
 * @author: lipei
 * @Date:2020-03-10 22:03
 */
public class BatchTableDemo {
    public static void main(String[] args) {
        //获取运行环境
        ExecutionEnvironment fbEnv = ExecutionEnvironment.getExecutionEnvironment();

        //创建一个tableEnvironment
        BatchTableEnvironment fbTableEnv = BatchTableEnvironment.create(fbEnv);


        //读取数据源
        DataSet<String> ds1 = fbEnv.readTextFile("src/file/text01.txt");

        //数据转换
        DataSet<Tuple2<String, String>> ds2 = ds1.map(new MapFunction<String, Tuple2<String, String>>() {
            private static final long serialVersionUID = -3027796541526131219L;

            @Override
            public Tuple2<String, String> map(String s) throws Exception {
                String[] splits =  s.split(",");
                return new Tuple2<>(splits[0], splits[1]);
            }
        });

        //DataSet 转table, 指定字段名
        Table table = fbTableEnv.fromDataSet(ds2, "id,name");


        Table table02 = table.select("name");

        //将表转换DataSet
        DataSet<String> ds3  = fbTableEnv.toDataSet(table02, String.class);

        try {
            ds3.print();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}

• SQL API

Flink 支持直接通过SQL操作数据

/**
 * @author: lipei
 * @Date:2020-03-10 22:03
 */
public class StreamSqlDemo {
    public static void main(String[] args) {
        //获取运行环境
        EnvironmentSettings fsSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();
        StreamExecutionEnvironment fsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        //创建一个tableEnvironment
        StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(fsEnv, fsSettings);

        //读取数据源
        DataStream<String> ds1 = fsEnv.readTextFile("src/file/text01.txt");

        //数据转换
        DataStream<Tuple2<String, String>> ds2 = ds1.map(new MapFunction<String, Tuple2<String, String>>() {
            private static final long serialVersionUID = -3027796541526131219L;

            @Override
            public Tuple2<String, String> map(String s) throws Exception {
                String[] splits =  s.split(",");
                return new Tuple2<>(splits[0], splits[1]);
            }
        });

        //DataStream 转sql, 指定字段名
        Table table = fsTableEnv.fromDataStream(ds2, "id,name");
        table.printSchema();

        //注册为一个表
        fsTableEnv.createTemporaryView("user01", table);

        Table table02 = fsTableEnv.sqlQuery("select * from user01").select("name");

        //将表转换DataStream
        DataStream<String> ds3  = fsTableEnv.toAppendStream(table02, String.class);
        ds3.print();
        try {
            fsEnv.execute("Flink stream sql");
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}

三、技术实现思路

3.1 需求背景

统计 所有售价小于10的订单明细信息

SELECT
    dajiangtai_orders.orderId,
    dajiangtai_orders.orderNo,
    dajiangtai_orders.userId,
    dajiangtai_orders.goodsId,
    dajiangtai_orders.goodsMoney,
    dajiangtai_goods.goodsName,
    dajiangtai_goods.sellingPrice,
    dajiangtai_orders.realTotalMoney,
    dajiangtai_orders.payFrom,
    dajiangtai_users.phoneNum,
    dajiangtai_users.userName,
    dajiangtai_users.address,
    dajiangtai_orders.createTime AS "orderCreateTime" 
FROM
    dajiangtai_orders 
    LEFT JOIN dajiangtai_goods  ON dajiangtai_orders.goodsId = dajiangtai_goods.goodsId
    LEFT JOIN dajiangtai_users  ON dajiangtai_orders.userId = dajiangtai_users.userId  
    where dajiangtai_goods.sellingPrice < 10

• 表映射关系

图片.png

3.2 分层设计

• DWD 层
  目的：获取数据明细表，对数据不进行任何过滤操作

SELECT
    dajiangtai_orders.orderId,
    dajiangtai_orders.orderNo,
    dajiangtai_orders.userId,
    dajiangtai_orders.goodsId,
    dajiangtai_orders.goodsMoney,
    dajiangtai_goods.goodsName,
    dajiangtai_goods.sellingPrice,
    dajiangtai_orders.realTotalMoney,
    dajiangtai_orders.payFrom,
    dajiangtai_users.phoneNum,
    dajiangtai_users.userName,
    dajiangtai_users.address,
    dajiangtai_orders.createTime AS "orderCreateTime" 
FROM
    dajiangtai_orders 
    LEFT JOIN dajiangtai_goods  ON dajiangtai_orders.goodsId = dajiangtai_goods.goodsId
    LEFT JOIN dajiangtai_users  ON dajiangtai_orders.userId = dajiangtai_users.userId  
    where dajiangtai_goods.sellingPrice < 10

• DM 层
  目的：在DWD 层上，进行字段和条数的过滤

SELECT
     *
FROM
    flink_order_detail
where  sellingPrice < 10
     

3.3 DWD 数据同步

3.3.1 源数据表insert 操作

a、找到受影响的子表Id
b、查询受影响的主表记录信息
c、进行insertOrUpdate操作

INSERT INTO `flink`.`dajiangtai_goods`(`goodsId`, `goodsName`, `sellingPrice`, `goodsStock`, `appraiseNum`) VALUES (1, '2', 20.00, 1, 1)；
a、字表  `flink`.`dajiangtai_goods`  受影响的rowKey 为 goodsId， id = 1
b、查询受影响的主表信息
SELECT
    dajiangtai_orders.orderId,
    dajiangtai_orders.orderNo,
    dajiangtai_orders.userId,
    dajiangtai_orders.goodsId,
    dajiangtai_orders.goodsMoney,
    dajiangtai_goods.goodsName,
    dajiangtai_goods.sellingPrice,
    dajiangtai_orders.realTotalMoney,
    dajiangtai_orders.payFrom,
    dajiangtai_users.phoneNum,
    dajiangtai_users.userName,
    dajiangtai_users.address,
    dajiangtai_orders.createTime AS "orderCreateTime" 
FROM
    dajiangtai_orders 
    LEFT JOIN dajiangtai_goods  ON dajiangtai_orders.goodsId = dajiangtai_goods.goodsId
    LEFT JOIN dajiangtai_users  ON dajiangtai_orders.userId = dajiangtai_users.userId  
    where dajiangtai_goods.goodsId in （1）
c、将b查询的记录同步到dwd库

3.3.1 源数据表update 操作

a、找到受影响的字表Id
b、查询受影响的主表记录信息
c、进行insertOrUpdate操作

• 例如：

update `flink`.`dajiangtai_goods` set sellingPrice = 25 where  `goodsId` = 1;
a、字表  `flink`.`dajiangtai_goods`  受影响的rowKey 为 goodsId， id = 1
b、查询受影响的主表信息
SELECT
    dajiangtai_orders.orderId,
    dajiangtai_orders.orderNo,
    dajiangtai_orders.userId,
    dajiangtai_orders.goodsId,
    dajiangtai_orders.goodsMoney,
    dajiangtai_goods.goodsName,
    dajiangtai_goods.sellingPrice,
    dajiangtai_orders.realTotalMoney,
    dajiangtai_orders.payFrom,
    dajiangtai_users.phoneNum,
    dajiangtai_users.userName,
    dajiangtai_users.address,
    dajiangtai_orders.createTime AS "orderCreateTime" 
FROM
    dajiangtai_orders 
    LEFT JOIN dajiangtai_goods  ON dajiangtai_orders.goodsId = dajiangtai_goods.goodsId
    LEFT JOIN dajiangtai_users  ON dajiangtai_orders.userId = dajiangtai_users.userId  
    where dajiangtai_goods.goodsId in （1）
c、将b查询的记录同步到dwd库

3.4 DW 数据同步
DW 层可以按照3.3 的步骤进行数据同步，直接落库，也可以基于DWD层直接进行where 查询