工具域优惠券领取事务事实表

主要任务

读取优惠券领用数据，写入Kafka 优惠券领用主题

思路分析

用户领取优惠券后，业务数据库的优惠券领用表会新增一条数据，因此操作类型为insert 的数据即为优惠券领取数据。

图解

代码编写

https://gitee.com/luan_hao/gmall-flink/blob/master/gmall-realtime/src/main/java/com/apache/gmall/app/dwd/db/DwdToolCouponGet.java

测试

bin/kafka-topics.sh --zookeeper hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka  --create --replication-factor 1 --partitions 1 --topic dwd_tool_coupon_get

bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092  --topic dwd_tool_coupon_get

开始启动

工具域优惠券使用（下单）事务事实表

主要任务

读取优惠券领用表数据，筛选优惠券下单数据，写入Kafka 优惠券下单主题。

思路分析

用户使用优惠券下单时，优惠券领用表的using_time 字段会更新为下单时间，同时coupon_status 字段会由 1401 更改为1402，因此优惠券下单数据应满足三个条件：

① 操作类型为 update；

② 当前 coupon_status 字段的值为1402；

③ 修改了 coupon_status 字段。

2**）执行步骤**

（1）通过三个条件筛选优惠券使用（下单）数据。

（2）建立Upsert-Kafka 表，将优惠券使用（下单）数据写入Kafka 优惠券使用（下单）事实主题。

图解

代码编写

https://gitee.com/luan_hao/gmall-flink/blob/master/gmall-realtime/src/main/java/com/apache/gmall/app/dwd/db/DwdToolCouponOrder.java

测试

bin/kafka-topics.sh --zookeeper hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka  --create --replication-factor 1 --partitions 1 --topic dwd_tool_coupon_order

bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092  --topic dwd_tool_coupon_order

开始启动

工具域优惠券使用（支付）事务事实表

主要任务

读取优惠券领用表数据，筛选优惠券支付数据，写入Kafka 优惠券支付主题。

思路分析

用户使用优惠券支付时，优惠券领用表的used_time 字段会更新为支付时间，因此优惠券支付数据应满足两个条件：

（1）操作类型为 update；

（2）修改了 used_time 字段。使用优惠券支付后，优惠券领用表数据就不会再发生变化，所以在操作类型为update 的前提下，只要used_time 不为null，就可以断定本次操作修改的是 used_time 字段。

图解

代码编写

https://gitee.com/luan_hao/gmall-flink/blob/master/gmall-realtime/src/main/java/com/apache/gmall/app/dwd/db/DwdToolCouponPay.java

测试

bin/kafka-topics.sh --zookeeper hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka  --create --replication-factor 1 --partitions 1 --topic dwd_tool_coupon_pay

bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092  --topic dwd_tool_coupon_pay

开始启动