营销型网站建设 价格,杏坛网站设计,室内装修设计自学软件,本地网站更新不了 vps登陆可以交易域支付成功事务事实表
从topic_db业务数据中筛选支付成功的数据从dwd_trade_order_detail主题中读取订单事实数据、LookUp字典表关联三张表形成支付成功宽表写入 Kafka 支付成功主题
执行步骤
设置ttl#xff0c;通过Interval join实现左右流的状态管理获取下单明细数据…交易域支付成功事务事实表
从topic_db业务数据中筛选支付成功的数据从dwd_trade_order_detail主题中读取订单事实数据、LookUp字典表关联三张表形成支付成功宽表写入 Kafka 支付成功主题
执行步骤
设置ttl通过Interval join实现左右流的状态管理获取下单明细数据用户必然要先下单才有可能支付成功因此支付成功明细数据集必然是订单明细数据集的子集。要注意Interval Join要求表中均为Append数据即“只能新增不能修改”订单明细表数据生成过程中用到了left join生成了回撤流看似不满足Interval Join的条件。但是回撤数据进入Kafka会以null值形式存在如果用Kafka Connector将订单明细封装为动态表null值会被过滤最终得到的是相同主键存在重复数据的Append流动态表本质上就是流满足Interval Join的条件。 Interval join只支持事件时间因此数据必须携带水位线建表时水位线的相关语法为 water for order_time as order_time - interval 5 second这里要求数据是timestamp(3)原有的时间数据类型是bigint类型的ts使用row_time as TO_TIMESTAMP_LTZ(ts,3)这个函数即可将原有的时间数据转换为水位线所需的数据类型 筛选支付数据 支付状态为支付成功操作类型为update 构建 LookUp 字典表联上述三张表形成支付成功宽表写入 Kafka 支付成功主题
核心代码如下 public void handle(StreamExecutionEnvironment env, TableEnvironment tableEnv, String groupId) {//核心业务逻辑//1. 读取TopicDB主题数据createTopicDb(groupId,tableEnv);//2. 筛选支付成功的数据从业务数据topic_db中filterPaymentTable(tableEnv);//3. 读取下单详情表数据, 从kafka读取数据createOrderDetailTable(tableEnv, groupId);//4. 创建base.dic字典表从HBase维度数据中读取createBaseDic(tableEnv);//tableEnv.executeSql(select * from order_detail).print();//tableEnv.executeSql(select * from base_dic).print();//tableEnv.executeSql(select to_timestamp_ltz(ts,3) from order_detail);//5. 使用interval join 完成支付成功流和订单详情数据关联intervalJoin(tableEnv);//6. 使用lookup join完成维度退化Table resultTable lookupJoin(tableEnv);//7. 创建upsert kafka连接器写出createKafkaSink(tableEnv);resultTable.insertInto(Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS).execute();}事实表动态分流 dwd层其他的事实表都是从topic_db中去业务数据库一张表的变更数据按照某些过滤后写入kafka的对应主题它们处理逻辑相似且较为简单可以结合配置表动态分流在同一个程序中处理。有点类似我们前面实现DIM层的动态配置。
清洗过滤和转换判断是否满足json格式如果满足转换为jsonObj对象读取配置表数据使用flink-cdc读取转换数据格式转换到对应bean对象中配置信息广播话然后跟主流数据进行连接筛选出需要的字段根据表中的sink table字段来动态写出到对应的kafka主题中
核心代码如下
public static void main(String[] args) {new DwdBaseDb().start(10019, 4, dwd_base_db, Constant.TOPIC_DB);}Overridepublic void handle(StreamExecutionEnvironment env, DataStreamSourceString stream) {//核心业务逻辑//1. 读取topic_db数据//stream.print();//2. 清洗过滤和转换, jsonObjStream是主流数据SingleOutputStreamOperatorJSONObject jsonObjStream filterJson(stream);//jsonObjStream.print();//3. 读取配置表数据使用flink-cdc读取,读取配置文件时并发度最好为1DataStreamSourceString tableProcessDwd getTableProcessDwd(env);//tableProcessDwd.print();4. 转换数据格式 string - TableProcessDwd - broadcastStream广播流数据SingleOutputStreamOperatorTableProcessDwd processDwdStream getProcessDwdStream(tableProcessDwd);MapStateDescriptorString, TableProcessDwd mapStateDescriptor new MapStateDescriptor(process_state, String.class, TableProcessDwd.class);BroadcastStreamTableProcessDwd broadcastStream processDwdStream.broadcast(mapStateDescriptor);//5. 连接主流和广播流对主流数据进行判断是否需要保留SingleOutputStreamOperatorTuple2JSONObject, TableProcessDwd processStream processBaseDb(jsonObjStream, broadcastStream, mapStateDescriptor);//processStream.print();//6. 筛选最后需要写出的字段SingleOutputStreamOperatorJSONObject dataStream filterColumns(processStream);//7. 通过sink_table的表名来动态写出到对应kafka主题//在setRecordSerializer()设置dataStream.sinkTo(FlinkSinkUtil.getKafkaSinkWithTopicName());}gitee地址 https://gitee.com/langpaian/gmall2023-realtime
