前言
今天开始交易域事实表的创建，上一节流量域中的表（其实就是一个 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；
