当前位置： 首页 > news >正文

Flink 实时数仓（五）【DWD 层搭建（三）交易域事实表】

news 来源：原创 2024/9/19 13:46:08

前言

        今天开始交易域事实表的创建，上一节流量域中的表（其实就是一个 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
SELECTdata['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
FROMtopic_db
WHERE`database` = 'gmall'
AND`table` = 'cart_info'
AND`type` = 'insert'
OR (`type` = 'update'AND'old'['sku_num'] is not nullANDCAST('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 关联加购事实表和字典表，为的就是对加购来源（比如广告/用户查询/促销活动/只能推荐）做一个维度退化：

SEELCTci.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
FROMcart_info ci
JOINbase_dic dic FOR SYSTEM_TOME AS OF ci.pt AS dic
ONci.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. 写出数据到 KafkatableEnv.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 开启checkpointenv.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；

相关文章：

  • 北京网站建设多少钱？
  • 辽宁网页制作哪家好_网站建设
  • 高端品牌网站建设_汉中网站制作
  • 数据结构与算法 - 优先级队列、阻塞队列
  • 我对于内存相关的三个问题的理解和总结——内存泄漏、内存溢出、野指针
  • 宏景eHR /ajax/ajaxService SQL注入漏洞复现
  • 【时时三省】unity test 测试框架 使用 code blocks 移植
  • 如何解决C#字典的线程安全问题
  • 40.组合综合Ⅱ
  • 【JavaEE精炼宝库】 网络编程套接字——UDP业务逻辑 | TCP流套接字编程及业务逻辑实现
  • 沉浸式企业VR展厅，重塑企业形象展示方式！
  • 程序员进阶架构知识体系、开发运维工具使用、Java体系知识扩展、前后端分离流程详解、设计模式开发实例汇总专栏分享
  • ComfyUI: 报EP Error错误（onnxruntime）
  • nginx的反向代理及负载均衡
  • 软件测试基础1--功能测试
  • EasyAR_稠密空间图
  • RIP综合练习
  • 在快速消费品业务中利用知识管理的指南
  • 2017-08-04 前端日报
  • axios请求、和返回数据拦截，统一请求报错提示_012
  • Docker入门（二） - Dockerfile
  • JAVA SE 6 GC调优笔记
  • javascript从右向左截取指定位数字符的3种方法
  • Java新版本的开发已正式进入轨道，版本号18.3
  • LeetCode算法系列_0891_子序列宽度之和
  • node.js
  • 看图轻松理解数据结构与算法系列(基于数组的栈)
  • 每个JavaScript开发人员应阅读的书【1】 - JavaScript: The Good Parts
  • 前端攻城师
  • 前端之Sass/Scss实战笔记
  • 区块链分支循环
  • 入门级的git使用指北
  • 文本多行溢出显示...之最后一行不到行尾的解决
  • 详解移动APP与web APP的区别
  • ​LeetCode解法汇总2808. 使循环数组所有元素相等的最少秒数
  • ​数据链路层——流量控制可靠传输机制 ​
  • ​香农与信息论三大定律
  • ​虚拟化系列介绍（十）
  • #AngularJS#$sce.trustAsResourceUrl
  • #define 用法
  • #gStore-weekly | gStore最新版本1.0之三角形计数函数的使用
  • #HarmonyOS：基础语法
  • #QT项目实战（天气预报）
  • $.extend({},旧的,新的)；合并对象，后面的覆盖前面的
  • （51单片机）第五章-A/D和D/A工作原理-A/D
  • （Oracle）SQL优化技巧（一）：分页查询
  • （规划）24届春招和25届暑假实习路线准备规划
  • (免费领源码）python#django#mysql校园校园宿舍管理系统84831-计算机毕业设计项目选题推荐
  • （十一）图像的罗伯特梯度锐化
  • （四）Linux Shell编程——输入输出重定向
  • （四十一）大数据实战——spark的yarn模式生产环境部署
  • (转)Oracle 9i 数据库设计指引全集(1)
  • （转）总结使用Unity 3D优化游戏运行性能的经验
  • （转载）PyTorch代码规范最佳实践和样式指南
  • . Flume面试题
  • ./configure、make、make install 命令
  • .net core + vue 搭建前后端分离的框架
  • .net core MVC 通过 Filters 过滤器拦截请求及响应内容