Flink 实时数仓（五）【DWD 层搭建（三）交易域事实表】_离线数仓订单dwd如何设计

前言

        今天开始交易域事实表的创建，上一节流量域中的表（其实就是一个 kafka 主题）数据来自于日志，而交易域的数据来自于业务系统，业务表之间是有关联性的。

        我们之前在离线数仓中（声明粒度（最细粒度） -> 确认维度（维度外键 + 退化字段） -> 确认事实（度量值）），往往都是会选择一个最细粒度的业务表作为主表（比如订单明细表，因为订单明细表的粒度是sku），然后和其他相关的业务表（比如订单表（订单表中可以获得订单的状态），订单明细活动关联表，订单明细优惠券关联表等）进行 join，保留维度外键，选择需要退化的字段，最后添加度量字段；

        那么，在事实表这里其实也是一样的，我们也不可避免需要对事实数据流之间进行 join，实时数据和维表之间的 join，所以我们现在需要熟悉一下 Flink 中如何使用 DataStream API 或者 Flink SQL 实现 join：

0、前置知识回顾

0.1、DataStream API  实现 join

        从官网可以看到，Flink 的窗口 API 分为两类：window join 和 interval join；我们可以分析一下在下面事实表的创建中应该选择哪种 join 方式：

• window join（官网并不推荐）
  • 滚动窗口：可能存在数据丢失（两条需要关联的数据刚好被窗口隔开）
  • 滑动窗口：可能存在数据重复
  • 会话窗口：可能存在窗口迟迟无法关闭，导致实时性变差
• interval join

关于 interval join，我们在学习 Flink 时间语义和水位线 的时候讲过：

间隔联结具体的定义方式是，我们给定两个时间点，分别叫作间隔的“上界”（upperBound）和“下界”（lowerBound）；于是对于一条流（不妨叫作 A）中的任意一个数据元素 a，就可以开辟一段时间间隔：[a.timestamp + lowerBound, a.timestamp + upperBound],即以 a 的时间戳为中心，下至下界点、上至上界点的一个闭区间：我们就把这段时间作为可以匹配另一条流数据的“窗口”范围。所以对于另一条流（不妨叫 B）中的数据元素 b，如果它的时间戳落在了这个区间范围内，a 和 b 就可以成功配对，进而进行计算输出结果。

所以匹配的条件为：a.timestamp + lowerBound <= b.timestamp <= a.timestamp + upperBound

这里需要注意：

• 做间隔联结的两条流 A 和 B，也必须基于相同的 key；
• 下界 lowerBound应该小于等于上界 upperBound，两者都可正可负；
• 间隔联结目前只支持事件时间语义。

0.2、Flink SQL 实现 join

这一块我们之前在学Flink SQL 查询的时候详细介绍过了，这里不多废话，回顾一下即可；

1、交易域加购事务事实表

        主要任务：提取订单表中的加购操作生成加购表，并将字典表中的相关维度退化到加购表中，写出到 Kafka 对应主题。

        前两节我们对日志数据的事实表进行构建的时候，考虑到日志表数据量大，于是我们将日志表根据不同类型拆分到了 5 个不同的 Kafka 主题当中。而这里我们处理业务系统中的数据并不需要那么做，毕竟数据量没那么大，这里我们只需要从 Maxwell 同步过来的 JSON 数据中过滤出我们需要的数据即可，不需要像日志数据那样多经过一次 Kafka ，影响效率；

1.1、实现思路

        事实表表需要通过和字典表进行 join 来做维度退化，所以这里我们需要使用 JDBC 连接器，将字典表中的数据封装成 Flink SQL 表；

        同样，我们需要用到 Kafka 连接器来将 Kafka 中所有的数据封装为一个 Flink SQL 表；

这个交易域加购事务事实表中，需要用到的业务表只有购物车表和订单表（用于对商品来源做一个维度退化，因为购物车表中只有来源 id，没有来源 value）；

1.2、代码实现

1.2.1、topic_db 表结构设计

首先，我们需要考虑如何将 Kafka 的 topic_db 中的数据保存到 Flink SQL 表中去？因为 topic_db 中存储的是 Maxwell 同步过来的 46 张表 JSON 格式的数据：

{
 "database":"gmall-211126-flink",
 "table":"base_trademark",
 "xid" : 188,
 "commit" : true,
 "type":"update",
 "ts":1652499295,
 "data":{
     "id":1,
     "tm_name":"三星",
     "icon_url":"/bbb/cccc"
     },
 "old":{
     "tm_name":"小米"
     }
}

        所以，我们需要考虑如何设计表结构？这里的 ts 字段并没有必要保留（ts 是 Maxwell 给加的，而且业务表中一般都有 create_time 和 operator_time 等字段），还有 xid、commit 等字段也不需要（都是 Maxwell 加的，对后面分析没有用）；

最后我们的表结构大概就是这样：

CREATE TABLE topic_db(
    `database` STRING,
    `table` STRING,
    `type` STRING,
    `data` MAP<STRING,STRING>,
    `old` MAP<STRING,STRING>,
    `pt` AS PROCTIME()
) WITH (
    'connector' = 'kafka',
    'topic' = 'topic_db',
    'properties.bootstrap.servers' = 'hadoop102:9092',
    'properties,group.id' = 'cartAdd',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json'
)

对于这里面的 data 和 old 是 JSON 对象类型，为了考虑到以后可以直接操作而不需要再做转换，我们可以考虑使用 Flink SQL 中的复杂数据类型 Map；

此外，我们还增加了一个时间字段 pt，因为后面我们需要做 lookup join，它需要我们提供一个处理时间字段；

1.2.2、topic_db 表实现

首先，Flink Table API 的执行环境有一点变化：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1); // 生产环境中设置为kafka主题的分区数
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

其它代码比如设置检查点和状态后端这里省略；

然后，我们在之前的 MyKafkaUtil 中再封装两个方法用来创建建表语句：

     /**
     * Kafka-Source DDL 的配置信息
     *
     * @param topic   数据源主题
     * @param groupId 消费者组
     * @return with 配置信息
     */
    public static String getKafkaDDL(String topic, String groupId) {
        return " with ('connector' = 'kafka', " +
                " 'topic' = '" + topic + "'," +
                " 'properties.bootstrap.servers' = '" + KAFKA_SERVER + "', " +
                " 'properties.group.id' = '" + groupId + "', " +
                " 'format' = 'json', " +
                " 'scan.startup.mode' = 'group-offsets')";
    }
 
    /**
     * topic_db 建表语句
     * @param groupId 消费者组
     * @return 完整的建表语句
     */
    public static String getTopicDb(String groupId) {
        return "CREATE TABLE topic_db(\n" +
                "    `database` STRING,\n" +
                "    `table` STRING,\n" +
                "    `type` STRING,\n" +
                "    `data` MAP<STRING,STRING>,\n" +
                "    `old` MAP<STRING,STRING>,\n" +
                "    `pt` AS PROCTIME()\n" +
                ")" + getKafkaDDL("topic_db",groupId);
    }

执行 DDL ：

        // TODO 2. 使用 DDL 方式读取 topic_db 创建表
        tableEnv.executeSql(MyKafkaUtil.getTopicDb("cart_add"));

        这样，我们的 topic_db 表就通过 Kafka 连接器和 Table API 创建完成了，之后我们就可以直接使用 Flink SQL 去查询我们想要的表的变更数据了；

1.2.3、过滤出加购数据

业务表中加购只两种情况，在 Maxwell 传递过来的数据中体现出来就是：

• type 字段为 insert
  • date 中的 sku_num 即为该次加购操作的 sku_num
• type 字段值为 update 且 data[sku_num] > old[sku_num]
  • data[sku_num] - old[sku_num] 即为该次加购操作的 sku_num

SELECT
     data['id'] id,
     data['user_id'] user_id,
     data['sku_id'] sku_id,
     data['cart_price'] cart_price,
     if(`type`='insert',`data`['sku_num'],CAST(CAST(`data`['sku_num'] AS INT) - CAST(`old`['sku_num'] AS INT)) AS STRING)) sku_num,
     data['sku_name'] sku_name,
     data['is_checked'] is_checked,
     data['create_time'] create_time,
     data['operate_time'] operate_time,
     data['is_ordered'] is_ordered,
     data['order_time'] order_time,
     data['source_id'] source_id,
     data['source_type'] source_type,
     pt
FROM
     topic_db
WHERE
     `database` = 'gmall'
AND
     `table` = 'cart_info'
AND
     `type` = 'insert'
OR (
        `type` = 'update'
        AND
            'old'['sku_num'] is not null
        AND
            CAST('old'['sku_num'] AS INT) < CAST('data'['sku_num'] AS INT)
    )

        这样，我们就可以从 topic_db 中读取到加购数据了，这些字段全部来自于 cart_info，每行数据代表一个加购操作；

        // TODO 3. 过滤加购数据
        Table cartAddTable = tableEnv.sqlQuery("SELECT\n" +
        "     data['id'] id,\n" +
        "     data['user_id'] ,\n" +
        "     data['sku_id'],\n" +
        "     data['cart_price'],\n" +
        "     if(`type`='insert',`data`['sku_num'],CAST(CAST(`data`['sku_num'] AS INT) - CAST(`old`['sku_num'] AS INT)) AS STRING)) sku_num,\n" +
        "     data['sku_name'] sku_name,\n" +
        "     data['is_checked'] is_checked,\n" +
        "     data['create_time'] create_time,\n" +
        "     data['operate_time'] operate_time,\n" +
        "     data['is_ordered'] is_ordered,\n" +
        "     data['order_time'] order_time,\n" +
        "     data['source_id'] source_id,\n" +
        "     data['source_type'] source_type,\n" +
        "     pt\n" +
        "FROM\n" +
        "     topic_db\n" +
        "WHERE\n" +
        "     `database` = 'gmall'\n" +
        "AND\n" +
        "     `table` = 'cart_info'\n" +
        "AND\n" +
        "     `type` = 'insert'\n" +
        "OR (\n" +
        "        `type` = 'update'\n" +
        "        AND\n" +
        "            'old'['sku_num'] is not null\n" +
        "        AND\n" +
        "            CAST('old'['sku_num'] AS INT) < CAST('data'['sku_num'] AS INT)\n" +
        "    )");
        tableEnv.createTemporaryView("cart_info_table",cartAddTable);

这里我们把提取出来的加购表创建成了一张临时表，这样下面我们才能用 FlinkSQL 对它和其他表进行关联； 

1.2.4、创建 lookup 表

        我们需要对字典表（base_dic）创建一张 lookup 表，用于和事实表去做 join。这里我们直接抽象出一个工具类，毕竟之后还有可能有其它表会和事实表去关联：

public class MysqlUtil {
 
    public static String getBaseDicLookUpDDL() {
        return "create table `base_dic`(\n" +
                "`dic_code` string,\n" +
                "`dic_name` string,\n" +
                "`parent_code` string,\n" +
                "`create_time` timestamp,\n" +
                "`operate_time` timestamp,\n" +
                "primary key(`dic_code`) not enforced\n" +
                ")" + MysqlUtil.mysqlLookUpTableDDL("base_dic");
    }
 
    public static String mysqlLookUpTableDDL(String tableName) {
        return "WITH (\n" +
                "'connector' = 'jdbc',\n" +
                "'url' = 'jdbc:mysql://hadoop102:3306/gmall',\n" +
                "'table-name' = '" + tableName + "',\n" +
                "'lookup.cache.max-rows' = '10',\n" +
                "'lookup.cache.ttl' = '1 hour',\n" +
                "'username' = 'root',\n" +
                "'password' = '123456',\n" +
                "'driver' = 'com.mysql.cj.jdbc.Driver'\n" +
                ")";
    }
}

        // TODO 4. 读取 MySQL 的 base_dic 作为 lookup 表
        tableEnv.executeSql(MysqlUtil.getBaseDicLookUpDDL());

这样，业务系统中字典表的 lookup 表就创建好了；

1.2.5、关联加购事实表和字典表

用 SQL 关联加购事实表和字典表，为的就是对加购来源（比如广告/用户查询/促销活动/只能推荐）做一个维度退化：

SEELCT
        ci.id,
        ci.user_id,
        ci.sku_id,
        ci.cart_price,
        ci.sku_num,
        ci.sku_name,
        ci.is_checked,
        ci.create_time,
        ci.operate_time,
        ci.is_ordered,
        ci.order_time,
        dic.dic_name source_type_name,
        ci.source_id,
        dic.dic_code source_type_id
FROM
    cart_info ci
JOIN
    base_dic dic FOR SYSTEM_TOME AS OF ci.pt AS dic
ON
    ci.source_type = dic.dic_code

关联后的表我们同样创建一张 Flink 临时表：

tableEnv.createTemporaryView("cart_with_dic_table",cartAddWithDicTable);

1.2.6、创建 DWD 层加购事务事实表

        数据现在已经有了，就是上面关联后的 cart_with_dic_table，现在我们需要创建 DWD 层的加购事务事实表：

注意：这张表最终是要写入到 Kafka 的 dwd_trade_cart_add 主题中的，所以我们需要在建表的时候带上配置信息（只需要指定 topic，不需要 groupId，因为我们是作为生产者向主题中写入数据）。

在 MyKafkaUtil 中添加方法：

public static String getKafkaSinkDDL(String topic) {
        return "WITH ( " +
                "  'connector' = 'kafka', " +
                "  'topic' = '" + topic + "', " +
                "  'properties.bootstrap.servers' = '" + KAFKA_SERVER + "', " +
                "  'format' = 'json' " +
                ")";
    }

创建 dwd_trade_cart_add 表：

// TODO 6. 使用 DDL 方式创建加购事实表
        // 前面创建的表是查询结果(Flink表,不包含配置信息比如Kafka主题等),
        // 这张表是要写入到Kafka主题中,所以只需要指定主题(指定主题就意味着写入该表的数据被写入到Kafka),
        // 这张表是Flink表它不会被消费,所以不需要指定消费者组
        tableEnv.executeSql("CREATE TABLE dwd_trade_cart_add(\n" +
                "        `id` STRING,\n" +
                "        `user_id` STRING,\n" +
                "        `sku_id` STRING,\n" +
                "        `cart_price` STRING,\n" +
                "        `sku_num` STRING,\n" +
                "        `sku_name` STRING,\n" +
                "        `is_checked` STRING,\n" +
                "        `create_time` STRING,\n" +
                "        `operate_time` STRING,\n" +
                "        `is_ordered` STRING,\n" +
                "        `order_time` STRING,\n" +
                "        `source_type_name` STRING,\n" +
                "        `source_id` STRING,\n" +
                "        `source_type_id` STRING\n" +
                ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_trade_cart_add"));

 1.2.7、写出数据

这一步我们只需要直接把关联后的临时表数据写入到上面的 dwd_trade_cart_add 表中即可：

 
        // TODO 7. 写出数据到 Kafka
        tableEnv.executeSql("INSERT INTO dwd_trade_cart_add SELECT * FROM cart_with_dic_table");
 
        // TODO 8. 启动任务
        env.execute("DwdTradeCartAdd");

2、交易域订单预处理表

在离线数仓中，我们针对订单这一类型创建的是累积快照事实表：

        累计快照事实表是基于一个业务流程（区别于业务过程，业务过程指的是一个业务的原子操作，而业务流程是由多个有关联的业务过程组成的）中的多个关键业务过程联合处理而构建的事实表，如交易流程中的下单、支付、发货、确认收货业务过程。

        累积型快照事实表通常具有多个日期字段，每个日期对应业务流程中的一个关键业务过程（里程碑）比如下面的 下单日期  -> 支付日期  -> 发货日期 -> 收货日期。

        这里，我们经过分析，订单明细表和取消订单明细表的数据来源、表结构都相同，差别只在业务过程和过滤条件，为了减少重复计算（重复关联多张表的数据），我们可以将两张表公共的关联过程提取出来，形成订单预处理表。

        关联订单明细表、订单表、订单明细活动关联表、订单明细优惠券关联表四张事实业务表和字典表（维度业务表）形成订单预处理表，写入 Kafka 对应主题。

        预处理表中要保留订单表的 type 和 old 字段，用于过滤订单明细数据和取消订单明细数据。

注意，预处理表在写出的时候不能再使用之前普通的 Kafka 连接器了，而是应该使用 Upsert Kafka 连接器（支持回撤流）； 

2.1、实现思路

        先把大致的框架列出来，很多内容比如创建 topic_db 表和 base_dic 的 lookup 表可以直接复用：

public class DwdTradeOrderPreProcess {
    public static void main(String[] args) throws Exception {
        // TODO 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); // 生产环境中设置为kafka主题的分区数
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
 
        // 1.1 开启checkpoint
        env.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/s/ck");
        env.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2); // 设置最大共存的checkpoint数量
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000L)); // 固定频率重启: 尝试3次重启,每5s重启一次
 
        // 1.2 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
 
        // TODO 2. 使用 DDL 方式读取 topic_db 创建表
        tableEnv.executeSql(MyKafkaUtil.getTopicDb("order_pre_process"));
 
        // TODO 3. 过滤 订单明细表
 
        // TODO 4. 过滤 订单表
        // TODO 5. 过滤 订单明细活动关联表
        // TODO 6. 过滤 订单明细优惠券关联表
        // TODO 7. 创建 base_dic 的 lookup 表
        tableEnv.executeSql(MysqlUtil.getBaseDicLookUpDDL());
        // TODO 8. 关联这 5 张表
        // TODO 9. 创建 upsert-kafka 表
        // TODO 10. 数据写出
        // TODO 11. 启动任务
        env.execute("DwdTradeOrderPreProcess");
    }

2.1.1、过滤订单明细表

注意：订单明细表不同于订单表，订单表只有新增（type 只有 insert）！但是订单明细表会有订单的修改（type 有 insert，也有 update）；

除了明显用不到的比如 img_url 等字段，别的以防万一全部保留（此外，还有 topic_db 的 pt 字段也需要查出来，后面要和 lookup 表 join 使用）： 

        // TODO 3. 过滤 订单明细表
        Table orderDetailTable = tableEnv.sqlQuery("SELECT\n" +
                "        data['id'] id,\n" +
                "        data['user_id'] user_id,\n" +
                "        data['sku_id'] sku_id,\n" +
                "        data['sku_name'] sku_name,\n" +
                "        data['order_price'] order_price,\n" +
                "        data['sku_num'] sku_num,\n" +
                "        data['create_time'] create_time,\n" +
                "        data['source_type'] source_type,\n" +
                "        data['source_id'] source_id,\n" +
                "        data['split_total_amount'] split_total_amount,\n" +
                "        data['split_activity_amount'] split_activity_amount,\n" +
                "        data['split_coupon_amount'] split_coupon_amount,\n" +
                "        pt\n" +
                "FROM\n" +
                "      topic_db\n" +
                "WHERE\n" +
                "      `database` = 'gmall'\n" +
                "AND\n" +
                "      `table` = 'order_detail'\n");
        tableEnv.createTemporaryView("order_detail_table",orderDetailTable);

2.1.2、过滤订单表

        订单表粒度大，所以没有明细的数据（商品的增删改），但是订单表中有订单的状态变化！所以我们还应该保存 type 和 old 字段，方便我们识别订单状态的变化：

        // TODO 4. 过滤 订单表
        Table orderInfoTable = tableEnv.sqlQuery("SELECT\n" +
                "        data['id'] id,\n" +
                "        data['user_id'] user_id,\n" +
                "        data['province_id'] province_id,\n" +
                "        data['order_status'] order_status,\n" +
                "        data['operate_time'] operate_time,\n" +
                "        `type`,\n" +
                "        `old`,\n" +
                "        ts\n" +
                "FROM\n" +
                "      topic_db\n" +
                "WHERE\n" +
                "      `database` = 'gmall'\n" +
                "AND\n" +
                "      `table` = 'order_info'\n" +
                "AND (`type` = 'insert' OR `type` = 'update')");
        tableEnv.createTemporaryView("order_info_table",orderInfoTable);

2.1.3、过滤订单明细活动关联表

订单明细活动关联表只有新增，所以这里不需要对 type 进行过滤；

        // TODO 5. 过滤 订单明细活动关联表
        Table orderActivityTable = tableEnv.sqlQuery("SELECT\n" +
                "        data['order_detail_id'] order_detail_id,\n" +
                "        data['activity_id'] activity_id,\n" +
                "        data['activity_rule_id'] activity_rule_id\n" +
                "FROM\n" +
                "      topic_db\n" +
                "WHERE\n" +
                "      `database` = 'gmall'\n" +
                "AND\n" +
                "      `table` = 'order_detail_activity'");
        tableEnv.createTemporaryView("order_activity_table",orderActivityTable);

2.1.4、过滤订单明细优惠券表

同样订单明细优惠券表只有新增，不需要过滤 type：

Table orderDetailCoupon = tableEnv.sqlQuery("select\n" +
                "data['order_detail_id'] order_detail_id,\n" +
                "data['coupon_id'] coupon_id\n" +
                "from `topic_db`\n" +
                "where `table` = 'order_detail_coupon'\n");
        tableEnv.createTemporaryView("order_detail_coupon", orderDetailCoupon);

2.1.5、关联 5 张表

        // TODO 8. 关联这 5 张表
        Table resultTable = tableEnv.sqlQuery("SELECT \n" +
                "od.id,\n" +
                "od.order_id,\n" +
                "oi.user_id,\n" +
                "oi.order_status,\n" +
                "od.sku_id,\n" +
                "od.sku_name,\n" +
                "oi.province_id,\n" +
                "oa.activity_id,\n" +
                "oa.activity_rule_id,\n" +
                "oc.coupon_id,\n" +
                "date_format(od.create_time, 'yyyy-MM-dd') date_id,\n" +
                "od.create_time,\n" +
                "date_format(oi.operate_time, 'yyyy-MM-dd') operate_date_id,\n" +
                "oi.operate_time,\n" +
                "od.source_id,\n" +
                "od.source_type source_type_id,\n" +
                "dic.dic_name source_type_name,\n" +
                "od.sku_num,\n" +
                "od.split_original_amount,\n" +
                "od.split_activity_amount,\n" +
                "od.split_coupon_amount,\n" +
                "od.split_total_amount,\n" +
                "oi.`type`,\n" +
                "oi.`old`,\n" +
                "od.od_ts,\n" +
                "oi.oi_ts,\n" +
                "current_row_timestamp() row_op_ts\n" +
                "FROM\n" +
                "    order_detail_table od\n" +
                "JOIN\n" +
                "    order_info_table oi\n" +
                "ON\n" +
                "    od.order_id = oi.id\n" +
                "LEFT JOIN\n" +
                "    order_activity_table oa\n" +
                "ON\n" +
                "    oa.order_detail_id = od.id\n" +
                "LEFT JOIN\n" +
                "    order_coupon_table oc\n" +
                "ON\n" +
                "    oa.order_detail_id = oc.id\n" +
                "JOIN\n" +
                "    base_dic FOR SYSTEM_TIME AS OF od.pt AS dic\n" +
                "ON\n" +
                "    od.source_type = dic.dic_code");
        tableEnv.createTemporaryView("result_table", resultTable);

2.1.6、创建 Upsert Kafka 表

        实时数据流在 left join 的过程中很可能出现数据的撤回，而之前我们使用的普通 Kafka 连接器是仅追加模式的，如果想要将有更新操作的结果表写入到 Kafka，就需要使用 Upsert Kafka 连接器了：

在 MyKafkaUtil 中创建方法（用来给建表时，添加 Kafka 参数）：

public static String getUpsertKafkaDDL(String topic) {
        return "WITH ( " +
                "  'connector' = 'upsert-kafka', " +
                "  'topic' = '" + topic + "', " +
                "  'properties.bootstrap.servers' = '" + KAFKA_SERVER + "', " +
                "  'key.format' = 'json', " +
                "  'value.format' = 'json' " +
                ")";
    }

创建表格，同样相当于是生产者，需要配置 Kafka 参数（topic，k，v类型）： 

        // TODO 9. 建立 Upsert-Kafka dwd_trade_order_pre_process 表
        tableEnv.executeSql("" +
                "create table dwd_trade_order_pre_process(\n" +
                "id string,\n" +
                "order_id string,\n" +
                "user_id string,\n" +
                "order_status string,\n" +
                "sku_id string,\n" +
                "sku_name string,\n" +
                "province_id string,\n" +
                "activity_id string,\n" +
                "activity_rule_id string,\n" +
                "coupon_id string,\n" +
                "date_id string,\n" +
                "create_time string,\n" +
                "operate_date_id string,\n" +
                "operate_time string,\n" +
                "source_id string,\n" +
                "source_type string,\n" +
                "source_type_name string,\n" +
                "sku_num string,\n" +
                "split_original_amount string,\n" +
                "split_activity_amount string,\n" +
                "split_coupon_amount string,\n" +
                "split_total_amount string,\n" +
                "`type` string,\n" +
                "`old` map<string,string>,\n" +
                "od_ts string,\n" +
                "oi_ts string,\n" +
                "row_op_ts timestamp_ltz(3),\n" +
                "primary key(id) not enforced\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_order_pre_process"));

2.1.7、写出到 Kafka

        // TODO 10. 数据写出
        tableEnv.executeSql("INSERT INTO dwd_trade_order_pre_process SELECT * FROM result_table");
 
        // TODO 11. 启动任务
        env.execute("DwdTradeOrderPreProcess");

总结

今天创建了 1 张事实表和一张辅助表（优化之后查询订单相关表的查询效率）

对于交易域加购事务事实表来说，需要根据公司业务逻辑来确定逻辑，比如有的公司的加购操作就是新增或者更新操作（更新商品的件数），但是有的公司的加购操作都是新增；

        我们创建了 topic_db 这张包含了全部业务系统 binlog 的数据表，并从中过滤出出了购物车相关的数据（type=insert 或者 update 的）。得到加购数据之后，我们需要对加购表中的加购来源字段（source_id）进行维度退化处理，这就需要去关联 base_dic 字典表。于是我们创建了 base_dic 的 lookup 表；

对于交易域订单预处理表，它并不是一个事实表，我们创建它是因为：和订单相关的业务过程（下单和取消订单）都需要从 订单明细表、订单表、订单明细活动关联表，订单明细优惠券关联表等去做关联，所以为了减少重复计算，我们可以这两张表的公共关联部分提取出来；创建的预处理表需要保留订单表的 type 和 old 字段（因为订单表的粒度是一个订单，它的变化就是订单状态的变化），所以为了方便后面过滤订单明细数据和取消订单明细数据（根据 old 中 order_status 字段的值和数据中 order_status 值的变化进行判断），

        和离线数仓一样，我们最终得到的预处理表的粒度也是最细粒度（订单明细表的粒度），一行代表一个商品（因为我们把订单明细表作为主表，并且和订单表进行了 join），同时具有丰富的维度属性（对订单活动、订单优惠券表进行了 left join）；

        最终得到的预处理表中，包含订单表的订单id、省份id、用户id、订单状态、操作时间、type 和 old 等字段；活动表的活动id和活动规则（维度退化）；优惠券表的优惠券id；