devbox
小舞很执着
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flink实时数仓_flink实时数仓项目

作者：小舞很执着 | 2024-08-07 15:02:21

赞

踩

flink实时数仓项目
  1. 相关数据：
  2. 启动日志：
  3. {"common":{"ar":"310000","ba":"Redmi","ch":"xiaomi","is_new":"1","md":"Redmi k30","mid":"mid_7","os":"Android 11.0","uid":"23","vc":"v2.1.111"},"start":{"entry":"icon","loading_time":13312,"open_ad_id":13,"open_ad_ms":9203,"open_ad_skip_ms":8503},"ts":1690869978000}
  4. 页面日志/曝光日志：
  5. {"actions":[{"action_id":"get_coupon","item":"3","item_type":"coupon_id","ts":1690869987153}],"common":{"ar":"310000","ba":"Redmi","ch":"xiaomi","is_new":"1","md":"Redmi k30","mid":"mid_7","os":"Android 11.0","uid":"23","vc":"v2.1.111"},"displays":[{"display_type":"recommend","item":"2","item_type":"sku_id","order":1,"pos_id":1},{"display_type":"promotion","item":"9","item_type":"sku_id","order":2,"pos_id":5},{"display_type":"promotion","item":"6","item_type":"sku_id","order":3,"pos_id":5},{"display_type":"promotion","item":"10","item_type":"sku_id","order":4,"pos_id":5},{"display_type":"query","item":"9","item_type":"sku_id","order":5,"pos_id":4}],"page":{"during_time":18307,"item":"10","item_type":"sku_id","last_page_id":"good_list","page_id":"good_detail","source_type":"activity"},"ts":1690869978000}
  6. {"actions":[{"action_id":"cart_minus_num","item":"2","item_type":"sku_id","ts":1690869984446}],"common":{"ar":"310000","ba":"Redmi","ch":"xiaomi","is_new":"1","md":"Redmi k30","mid":"mid_7","os":"Android 11.0","uid":"23","vc":"v2.1.111"},"page":{"during_time":12892,"last_page_id":"good_detail","page_id":"cart"},"ts":1690869978000}
  7. {"common":{"ar":"310000","ba":"Redmi","ch":"xiaomi","is_new":"1","md":"Redmi k30","mid":"mid_7","os":"Android 11.0","uid":"23","vc":"v2.1.111"},"page":{"during_time":10393,"item":"1,9","item_type":"sku_ids","last_page_id":"cart","page_id":"trade"},"ts":1690869978000}

017-采集模块-日志数据采集之SpringBoot创建项目&加参数测试

019-采集模块-日志数据采集之数据落盘&写入Kafka  本地测试

  1. 启动zookeeper:
  2.    bin/zkServer.sh start 
  3.  
  4. 启动kafaka:
  5.    启动kafka:
  6.    bin/kafka-server-start.sh config/server9092.properties 
  7.    
  8.    启动消费者"ods_base_log":
  9.    bin/kafka-console-consumer.sh --bootstrap-server  bigdata- 
  10.    training01.erongda.com:9092 -from-beginning --topic  ods_base_log
  11.  
  12. 启动GmallLoggerApplication.java 
  13.  
  14. 启动：
  15. java -jar gma112020-mock-1og-2020-12-18.iar
logback.xml 添加"将某一个包下日志单独打印日志"控制台不会打印出启动日志 
  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <configuration>
  3.     <property name="LOG_HOME" value="/opt/modules/gmall-flink/rt_applog/logs"/>
  4.     <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
  5.         <encoder>
  6.             <pattern>%msg%n</pattern>
  7.         </encoder>
  8.     </appender>
  9.  
  10.     <appender name="rollingFile" class="ch.qos.logback.core.rolling.RollingFileAppender">
  11.         <file>${LOG_HOME}/app.log</file>
  12.         <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
  13.             <fileNamePattern>${LOG_HOME}/app.%d{yyyy-MM-dd}.log</fileNamePattern>
  14.         </rollingPolicy>
  15.         <encoder>
  16.             <pattern>%msg%n</pattern>
  17.         </encoder>
  18.     </appender>
  19.  
  20.     <!-- 将某一个包下日志单独打印日志 -->
  21.     <logger name="com.atguigu.gmalllogger.controller.LoggerController"
  22.             level="INFO" additivity="false">
  23.         <appender-ref ref="rollingFile"/>
  24.         <appender-ref ref="console"/>
  25.     </logger>
  26.  
  27.     <root level="error" additivity="false">
  28.         <appender-ref ref="console"/>
  29.     </root>
  30. </configuration>

 java -jar gma112020-mock-1og-2020-12-18.iar

 控制台数据有了：

 kafka中ods_base_log数据有了：

020-采集模块-日志数据采集之数据落盘&写入Kafka  单机测试

  1. 启动zookeeper:
  2.    bin/zkServer.sh start 
  3.  
  4. 启动kafaka:
  5.    启动kafka:
  6.    bin/kafka-server-start.sh config/server9092.properties 
  7.    
  8.    启动消费者"ods_base_log":
  9.    bin/kafka-console-consumer.sh --bootstrap-server  bigdata- 
  10.    training01.erongda.com:9092 -from-beginning --topic  ods_base_log
  11.  
  12.  
  13. 启动：
  14. java -jar gma112020-mock-1og-2020-12-18.iar
  15. java -jar gmall-logger.jar

 

029-采集模块-业务数据采集之FlinkCDC  DataStream方式测试

  1. package com.atguigu;
  2.  
  3. import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
  4. import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
  5. import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
  6. import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
  7. import org.apache.flink.runtime.state.filesystem.FsStateBackend;
  8. import org.apache.flink.streaming.api.CheckpointingMode;
  9. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  11.  
  12. public class FlinkCDCWithCustomerDeserialization {
  13.  
  14.     public static void main(String[] args) throws Exception {
  15.  
  16.         //1.获取执行环境
  17.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  18.         env.setParallelism(1);
  19.  
  20.         //2.通过FlinkCDC构建SourceFunction并读取数据
  21.         DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
  22.                 .hostname("127.0.0.1")
  23.                 .port(3306)
  24.                 .username("root")
  25.                 .password("123456")
  26.                 .databaseList("gmall-210325-flink")
  27.                 .tableList("gmall-210325-flink.base_trademark")   //如果不添加该参数,则消费指定数据库中所有表的数据.如果指定,指定方式为db.table
  28.                 .deserializer(new StringDebeziumDeserializationSchema())
  29.                 .startupOptions(StartupOptions.initial())
  30.                 .build();
  31.         DataStreamSource<String> streamSource = env.addSource(sourceFunction);
  32.  
  33.         //3.打印数据
  34.         streamSource.print();
  35.  
  36.         //4.启动任务
  37.         env.execute("FlinkCDCWithCustomerDeserialization");
  38.  
  39.     }
  40.  
  41. }

 

 

030-采集模块-FlinkCDC  DataStreamAPI  设置CK&打包&开启集群

032-采集模块-业务数据采集之FlinkCDC  FlinkSQLAPI  测试.mp4

  1. package com.atguigu;
  2.  
  3. import org.apache.flink.api.java.tuple.Tuple2;
  4. import org.apache.flink.streaming.api.datastream.DataStream;
  5. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  6. import org.apache.flink.table.api.Table;
  7. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  8. import org.apache.flink.types.Row;
  9.  
  10. public class FlinkCDCWithSQL {
  11.  
  12.     public static void main(String[] args) throws Exception {
  13.  
  14.         //1.获取执行环境
  15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  16.         env.setParallelism(1);
  17.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
  18.  
  19.         //2.DDL方式建表
  20.         tableEnv.executeSql("CREATE TABLE mysql_binlog ( " +
  21.                 " id STRING NOT NULL, " +
  22.                 " tm_name STRING, " +
  23.                 " logo_url STRING " +
  24.                 ") WITH ( " +
  25.                 " 'connector' = 'mysql-cdc', " +
  26.                 " 'hostname' = '127.0.0.1', " +
  27.                 " 'port' = '3306', " +
  28.                 " 'username' = 'root', " +
  29.                 " 'password' = '123456', " +
  30.                 " 'database-name' = 'gmall-210325-flink', " +
  31.                 " 'table-name' = 'base_trademark' " +
  32.                 ")");
  33.  
  34.         //3.查询数据
  35.         Table table = tableEnv.sqlQuery("select * from mysql_binlog");
  36.  
  37.         //4.将动态表转换为流
  38.         DataStream<Tuple2<Boolean, Row>> retractStream = tableEnv.toRetractStream(table, Row.class);
  39.         retractStream.print();
  40.  
  41.         //5.启动任务
  42.         env.execute("FlinkCDCWithSQL");
  43.  
  44.     }
  45.  
  46. }

  1. 设置checkpoint：
  2. enableCheckpoint:上一次头和下一次头间隔的时间 生产环境为5min    
  3. setCheckpointTime:超时时间为10000s 具体看生产环境中状态保存的时间,如果是5秒保存状态就需要设置为10s
  4. setMaxConcurrentCheckpoint:2 最多可以存在几个checkpoint
  5. setMinPauseBetweenCheckpoint:3000s 上一次头和下一次尾的间隔时间
  6. setRestartStrategy:(3,5)如果无法重启最多可以重启3次,每次间隔5s 注意：老版本需要设置,新版本不需要(新版本设置比较合理) 重启策越
  7. 1.10 默认重启int的最大值,所以需要配置(不然一直会重启) 生产环境默认就可以 如果三次都重启失败,任务就失败

修改序列化：

  1. package com.atguigu;
  2.  
  3. import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
  4. import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
  5. import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
  6. import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
  7. import org.apache.flink.runtime.state.filesystem.FsStateBackend;
  8. import org.apache.flink.streaming.api.CheckpointingMode;
  9. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  11.  
  12. public class FlinkCDCWithCustomerDeserialization {
  13.  
  14.     public static void main(String[] args) throws Exception {
  15.  
  16.         //1.获取执行环境
  17.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  18.         env.setParallelism(1);
  19.  
  20.         //2.通过FlinkCDC构建SourceFunction并读取数据
  21.         DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
  22.                 .hostname("127.0.0.1")
  23.                 .port(3306)
  24.                 .username("root")
  25.                 .password("123456")
  26.                 .databaseList("gmall-210325-flink")
  27.                 .tableList("gmall-210325-flink.base_trademark")   //如果不添加该参数,则消费指定数据库中所有表的数据.如果指定,指定方式为db.table
  28.                 .deserializer(new CustomerDeserialization())  //自定义序列化
  29.                 //.deserializer(new StringDebeziumDeserializationSchema()) //默认序列化
  30.                 .startupOptions(StartupOptions.initial())
  31.                 .build();
  32.         DataStreamSource<String> streamSource = env.addSource(sourceFunction);
  33.  
  34.         //3.打印数据
  35.         streamSource.print();
  36.  
  37.         //4.启动任务
  38.         env.execute("FlinkCDCWithCustomerDeserialization");
  39.  
  40.     }
  41.  
  42. }

 

  1. package com.atguigu;
  2.  
  3. import com.alibaba.fastjson.JSONObject;
  4. import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
  5. import io.debezium.data.Envelope;
  6. import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
  7. import org.apache.flink.api.common.typeinfo.TypeInformation;
  8. import org.apache.flink.util.Collector;
  9. import org.apache.kafka.connect.data.Field;
  10. import org.apache.kafka.connect.data.Schema;
  11. import org.apache.kafka.connect.data.Struct;
  12. import org.apache.kafka.connect.source.SourceRecord;
  13.  
  14. import java.util.List;
  15.  
  16. public class CustomerDeserialization implements DebeziumDeserializationSchema<String> {
  17.  
  18.     /**
  19.      * 封装的数据格式
  20.      * {
  21.      * "database":"",
  22.      * "tableName":"",
  23.      * "before":{"id":"","tm_name":""....},
  24.      * "after":{"id":"","tm_name":""....},
  25.      * "type":"c u d",
  26.      * //"ts":156456135615
  27.      * }
  28.      */
  29.     @Override
  30.     public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
  31.  
  32.         //1.创建JSON对象用于存储最终数据
  33.         JSONObject result = new JSONObject();
  34.  
  35.         //2.获取库名&表名
  36.         String topic = sourceRecord.topic();
  37.         String[] fields = topic.split("\\.");
  38.         String database = fields[1];
  39.         String tableName = fields[2];
  40.  
  41.         Struct value = (Struct) sourceRecord.value();
  42.         //3.获取"before"数据
  43.         Struct before = value.getStruct("before");
  44.         JSONObject beforeJson = new JSONObject();
  45.         if (before != null) {
  46.             Schema beforeSchema = before.schema();
  47.             List<Field> beforeFields = beforeSchema.fields();
  48.             for (Field field : beforeFields) {
  49.                 Object beforeValue = before.get(field);
  50.                 beforeJson.put(field.name(), beforeValue);
  51.             }
  52.         }
  53.  
  54.         //4.获取"after"数据
  55.         Struct after = value.getStruct("after");
  56.         JSONObject afterJson = new JSONObject();
  57.         if (after != null) {
  58.             Schema afterSchema = after.schema();
  59.             List<Field> afterFields = afterSchema.fields();
  60.             for (Field field : afterFields) {
  61.                 Object afterValue = after.get(field);
  62.                 afterJson.put(field.name(), afterValue);
  63.             }
  64.         }
  65.  
  66.         //5.获取操作类型  CREATE UPDATE DELETE
  67.         Envelope.Operation operation = Envelope.operationFor(sourceRecord);
  68.         String type = operation.toString().toLowerCase();
  69.         if ("create".equals(type)) {
  70.             type = "insert";
  71.         }
  72.  
  73.         //6.将字段写入JSON对象
  74.         result.put("database", database);
  75.         result.put("tableName", tableName);
  76.         result.put("before", beforeJson);
  77.         result.put("after", afterJson);
  78.         result.put("type", type);
  79.  
  80.         //7.输出数据
  81.         collector.collect(result.toJSONString());
  82.  
  83.     }
  84.  
  85.     @Override
  86.     public TypeInformation<String> getProducedType() {
  87.         return BasicTypeInfo.STRING_TYPE_INFO;
  88.     }
  89. }
  1. 比较FlinkCDC:
  2.          DataStream:
  3.                  优点：多库多表
  4.                  缺点：需要自定义反序列化器(灵活)
  5.          FlinkSQL:
  6.                  优点：不需要自定义反序列化器
  7.                  缺点：单表查询(Flinkcdc可以通过参数传给bean)

035--采集模块-业务数据采集之FlinkCDC  DataStreamAPI  自定义反序列化器  代码测试

  1. 序列化前：
  2. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1693292078, file=mysql-bin.000071, pos=528, row=1, server_id=1, event=2}} ConnectRecord{topic='mysql_binlog_source.gmall-210325-flink.base_trademark', kafkaPartition=null, key=Struct{id=12}, keySchema=Schema{mysql_binlog_source.gmall_210325_flink.base_trademark.Key:STRUCT}, value=Struct{after=Struct{id=12,tm_name=test,logo_url=test},source=Struct{version=1.4.1.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1693292078000,db=gmall-210325-flink,table=base_trademark,server_id=1,file=mysql-bin.000071,pos=691,row=0,thread=6},op=c,ts_ms=1693292078631}, valueSchema=Schema{mysql_binlog_source.gmall_210325_flink.base_trademark.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  3. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1693292093, file=mysql-bin.000071, pos=843, row=1, server_id=1, event=2}} ConnectRecord{topic='mysql_binlog_source.gmall-210325-flink.base_trademark', kafkaPartition=null, key=Struct{id=12}, keySchema=Schema{mysql_binlog_source.gmall_210325_flink.base_trademark.Key:STRUCT}, value=Struct{before=Struct{id=12,tm_name=test,logo_url=test},after=Struct{id=12,tm_name=test,logo_url=tes},source=Struct{version=1.4.1.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1693292093000,db=gmall-210325-flink,table=base_trademark,server_id=1,file=mysql-bin.000071,pos=1006,row=0,thread=6},op=u,ts_ms=1693292093136}, valueSchema=Schema{mysql_binlog_source.gmall_210325_flink.base_trademark.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  4. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={ts_sec=1693292104, file=mysql-bin.000071, pos=1179, row=1, server_id=1, event=2}} ConnectRecord{topic='mysql_binlog_source.gmall-210325-flink.base_trademark', kafkaPartition=null, key=Struct{id=12}, keySchema=Schema{mysql_binlog_source.gmall_210325_flink.base_trademark.Key:STRUCT}, value=Struct{before=Struct{id=12,tm_name=test,logo_url=tes},source=Struct{version=1.4.1.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1693292104000,db=gmall-210325-flink,table=base_trademark,server_id=1,file=mysql-bin.000071,pos=1342,row=0,thread=6},op=d,ts_ms=1693292104741}, valueSchema=Schema{mysql_binlog_source.gmall_210325_flink.base_trademark.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  5.  
  6.  
  7. 序列化后：
  8. {"database":"gmall-210325-flink","before":{},"after":{"tm_name":"test","logo_url":"test","id":12},"type":"insert","tableName":"base_trademark"}
  9. {"database":"gmall-210325-flink","before":{"tm_name":"test","logo_url":"test","id":12},"after":{"tm_name":"test","logo_url":"tes","id":12},"type":"update","tableName":"base_trademark"}
  10. {"database":"gmall-210325-flink","before":{"tm_name":"test","logo_url":"tes","id":12},"after":{},"type":"delete","tableName":"base_trademark"}
  1. package com.atguigu;
  2.  
  3. import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
  4. import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
  5. import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
  6. import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
  7. import org.apache.flink.runtime.state.filesystem.FsStateBackend;
  8. import org.apache.flink.streaming.api.CheckpointingMode;
  9. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  11.  
  12. public class FlinkCDCWithCustomerDeserialization {
  13.  
  14.     public static void main(String[] args) throws Exception {
  15.  
  16.         //1.获取执行环境
  17.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  18.         env.setParallelism(1);
  19.  
  20.         //2.通过FlinkCDC构建SourceFunction并读取数据
  21.         DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
  22.                 .hostname("127.0.0.1")
  23.                 .port(3306)
  24.                 .username("root")
  25.                 .password("123456")
  26.                 .databaseList("gmall-210325-flink")
  27.                 .tableList("gmall-210325-flink.base_trademark")   //如果不添加该参数,则消费指定数据库中所有表的数据.如果指定,指定方式为db.table
  28.                 .deserializer(new CustomerDeserialization())  //自定义序列化
  29.                 //.deserializer(new StringDebeziumDeserializationSchema()) //默认序列化
  30.                 .startupOptions(StartupOptions.initial())
  31.                 .build();
  32.         DataStreamSource<String> streamSource = env.addSource(sourceFunction);
  33.  
  34.         //3.打印数据
  35.         streamSource.print();
  36.  
  37.         //4.启动任务
  38.         env.execute("FlinkCDCWithCustomerDeserialization");
  39.  
  40.     }
  41.  
  42. }
  1. package com.atguigu;
  2.  
  3. import com.alibaba.fastjson.JSONObject;
  4. import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
  5. import io.debezium.data.Envelope;
  6. import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
  7. import org.apache.flink.api.common.typeinfo.TypeInformation;
  8. import org.apache.flink.util.Collector;
  9. import org.apache.kafka.connect.data.Field;
  10. import org.apache.kafka.connect.data.Schema;
  11. import org.apache.kafka.connect.data.Struct;
  12. import org.apache.kafka.connect.source.SourceRecord;
  13.  
  14. import java.util.List;
  15.  
  16. public class CustomerDeserialization implements DebeziumDeserializationSchema<String> {
  17.  
  18.     /**
  19.      * 封装的数据格式
  20.      * {
  21.      * "database":"",
  22.      * "tableName":"",
  23.      * "before":{"id":"","tm_name":""....},
  24.      * "after":{"id":"","tm_name":""....},
  25.      * "type":"c u d",
  26.      * //"ts":156456135615
  27.      * }
  28.      */
  29.     @Override
  30.     public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
  31.  
  32.         //1.创建JSON对象用于存储最终数据
  33.         JSONObject result = new JSONObject();
  34.  
  35.         //2.获取库名&表名
  36.         String topic = sourceRecord.topic();
  37.         String[] fields = topic.split("\\.");
  38.         String database = fields[1];
  39.         String tableName = fields[2];
  40.  
  41.         Struct value = (Struct) sourceRecord.value();
  42.         //3.获取"before"数据
  43.         Struct before = value.getStruct("before");
  44.         JSONObject beforeJson = new JSONObject();
  45.         if (before != null) {
  46.             Schema beforeSchema = before.schema();
  47.             List<Field> beforeFields = beforeSchema.fields();
  48.             for (Field field : beforeFields) {
  49.                 Object beforeValue = before.get(field);
  50.                 beforeJson.put(field.name(), beforeValue);
  51.             }
  52.         }
  53.  
  54.         //4.获取"after"数据
  55.         Struct after = value.getStruct("after");
  56.         JSONObject afterJson = new JSONObject();
  57.         if (after != null) {
  58.             Schema afterSchema = after.schema();
  59.             List<Field> afterFields = afterSchema.fields();
  60.             for (Field field : afterFields) {
  61.                 Object afterValue = after.get(field);
  62.                 afterJson.put(field.name(), afterValue);
  63.             }
  64.         }
  65.  
  66.         //5.获取操作类型  CREATE UPDATE DELETE
  67.         Envelope.Operation operation = Envelope.operationFor(sourceRecord);
  68.         String type = operation.toString().toLowerCase();
  69.         if ("create".equals(type)) {
  70.             type = "insert";
  71.         }
  72.  
  73.         //6.将字段写入JSON对象
  74.         result.put("database", database);
  75.         result.put("tableName", tableName);
  76.         result.put("before", beforeJson);
  77.         result.put("after", afterJson);
  78.         result.put("type", type);
  79.  
  80.         //7.输出数据
  81.         collector.collect(result.toJSONString());
  82.  
  83.     }
  84.  
  85.     @Override
  86.     public TypeInformation<String> getProducedType() {
  87.         return BasicTypeInfo.STRING_TYPE_INFO;
  88.     }
  89. }

041--采集模块-业务数据采集之读取MySQL数据并写入Kafka  测试

启动zookeeper：

  1. bin/zkServer.sh start 
  2. 查看状态：
  3. bin/zkServer.sh status

 启动kafka：

  1. 创建消费者：
  2. bin/kafka-topics.sh --create --zookeeper bigdata-training01.erongda.com:2181/kafka --replication-factor 2 --partitions 3 --topic ods_base_db
  3.  
  4. 启动消费者：
  5. bin/kafka-console-consumer.sh --bootstrap-server  bigdata-training01.erongda.com:9092 -from-beginning --topic ods_base_db 
  1. package com.atguigu.app.ods;
  2.  
  3. import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
  4. import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
  5. import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
  6. import com.atguigu.app.function.CustomerDeserialization;
  7. import com.atguigu.utils.MyKafkaUtil;
  8. import org.apache.flink.runtime.state.filesystem.FsStateBackend;
  9. import org.apache.flink.streaming.api.CheckpointingMode;
  10. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  12.  
  13. public class FlinkCDC {
  14.  
  15.     public static void main(String[] args) throws Exception {
  16.  
  17.         //1.获取执行环境
  18.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  19.         env.setParallelism(1);
  20.  
  21.         //1.1 设置CK&状态后端
  22.         //env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/gmall-flink-210325/ck"));
  23.         //env.enableCheckpointing(5000L);
  24.         //env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
  25.         //env.getCheckpointConfig().setCheckpointTimeout(10000L);
  26.         //env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
  27.         //env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);
  28.  
  29.         //env.setRestartStrategy(RestartStrategies.fixedDelayRestart());
  30.  
  31.         //2.通过FlinkCDC构建SourceFunction并读取数据
  32.         DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
  33.                 .hostname("127.0.0.1")
  34.                 .port(3306)
  35.                 .username("root")
  36.                 .password("123456")
  37.                 .databaseList("gmall-210325-flink")
  38.                 .deserializer(new CustomerDeserialization())
  39.                 .startupOptions(StartupOptions.latest())
  40.                 .build();
  41.         DataStreamSource<String> streamSource = env.addSource(sourceFunction);
  42.  
  43.         //3.打印数据并将数据写入Kafka
  44.         streamSource.print();
  45.         String sinkTopic = "ods_base_db";
  46.         streamSource.addSink(MyKafkaUtil.getKafkaProducer(sinkTopic));
  47.  
  48.         //4.启动任务
  49.         env.execute("FlinkCDC");
  50.     }
  51.  
  52. }

 

 045-DWD&DIM-行为数据  将数据转换为JSON对象

  1. public class FlinkCDCWithCustomerDeserialization {
  2.  
  3.     public static void main(String[] args) throws Exception {
  4.  
  5.         //1.获取执行环境
  6.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  7.         env.setParallelism(1);
  8.  
  9.         //2.通过FlinkCDC构建SourceFunction并读取数据
  10.         DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
  11.                 .hostname("127.0.0.1")
  12.                 .port(3306)
  13.                 .username("root")
  14.                 .password("123456")
  15.                 .databaseList("gmall-210325-flink")
  16.                 .tableList("gmall-210325-flink.base_trademark")   //如果不添加该参数,则消费指定数据库中所有表的数据.如果指定,指定方式为db.table
  17.                 .deserializer(new CustomerDeserialization())  //自定义序列化
  18.                 //.deserializer(new StringDebeziumDeserializationSchema()) //默认序列化
  19.                 .startupOptions(StartupOptions.initial())
  20.                 .build();
  21.         DataStreamSource<String> streamSource = env.addSource(sourceFunction);
  22.  
  23.         //3.打印数据
  24.         streamSource.print();
  25.  
  26.         //4.启动任务
  27.         env.execute("FlinkCDCWithCustomerDeserialization");
  28.  
  29.     }
  30.  
  31. }

048-DWD&DIM-行为数据  

  1. 启动kafka:
  2. bin/kafka-server-start.sh config/server9092.properties 
  3.  
  4. 启动消费者：
  5. bin/kafka-console-consumer.sh --bootstrap-server  bigdata-training01.erongda.com:9092 -from-beginning --topic  dwd_start_log
  6.  
  7. bin/kafka-console-consumer.sh --bootstrap-server  bigdata-training01.erongda.com:9092 -from-beginning --topic  dwd_page_log
  8.  
  9.  
  10. bin/kafka-console-consumer.sh --bootstrap-server  bigdata-training01.erongda.com:9092 -from-beginning --topic  dwd_display_log
  11.  
  12. 启动生产者：
  13. bin/kafka-console-producer.sh --broker-list  bigdata-training01.erongda.com:9092 --topic ods_base_log
  14.  
  15.  
  16. 相关数据：
  17. 启动日志：
  18. {"common":{"ar":"310000","ba":"Redmi","ch":"xiaomi","is_new":"1","md":"Redmi k30","mid":"mid_7","os":"Android 11.0","uid":"23","vc":"v2.1.111"},"start":{"entry":"icon","loading_time":13312,"open_ad_id":13,"open_ad_ms":9203,"open_ad_skip_ms":8503},"ts":1690869978000}
  19. 页面日志/曝光日志：
  20. {"actions":[{"action_id":"get_coupon","item":"3","item_type":"coupon_id","ts":1690869987153}],"common":{"ar":"310000","ba":"Redmi","ch":"xiaomi","is_new":"1","md":"Redmi k30","mid":"mid_7","os":"Android 11.0","uid":"23","vc":"v2.1.111"},"displays":[{"display_type":"recommend","item":"2","item_type":"sku_id","order":1,"pos_id":1},{"display_type":"promotion","item":"9","item_type":"sku_id","order":2,"pos_id":5},{"display_type":"promotion","item":"6","item_type":"sku_id","order":3,"pos_id":5},{"display_type":"promotion","item":"10","item_type":"sku_id","order":4,"pos_id":5},{"display_type":"query","item":"9","item_type":"sku_id","order":5,"pos_id":4}],"page":{"during_time":18307,"item":"10","item_type":"sku_id","last_page_id":"good_list","page_id":"good_detail","source_type":"activity"},"ts":1690869978000}
  21. {"actions":[{"action_id":"cart_minus_num","item":"2","item_type":"sku_id","ts":1690869984446}],"common":{"ar":"310000","ba":"Redmi","ch":"xiaomi","is_new":"1","md":"Redmi k30","mid":"mid_7","os":"Android 11.0","uid":"23","vc":"v2.1.111"},"page":{"during_time":12892,"last_page_id":"good_detail","page_id":"cart"},"ts":1690869978000}
  22. {"common":{"ar":"310000","ba":"Redmi","ch":"xiaomi","is_new":"1","md":"Redmi k30","mid":"mid_7","os":"Android 11.0","uid":"23","vc":"v2.1.111"},"page":{"during_time":10393,"item":"1,9","item_type":"sku_ids","last_page_id":"cart","page_id":"trade"},"ts":1690869978000}
  1. //数据流：web/app -> Nginx -> SpringBoot -> Kafka(ods) -> FlinkApp -> Kafka(dwd)
  2. //程  序：mockLog -> Nginx -> Logger.sh  -> Kafka(ZK)  -> BaseLogApp -> kafka
  3. public class BaseLogApp {
  4.  
  5.     public static void main(String[] args) throws Exception {
  6.  
  7.         //TODO 1.获取执行环境
  8.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  9.         env.setParallelism(1);
  10.  
  11.         //1.1 设置CK&状态后端
  12.         //env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/gmall-flink-210325/ck"));
  13.         //env.enableCheckpointing(5000L);
  14.         //env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
  15.         //env.getCheckpointConfig().setCheckpointTimeout(10000L);
  16.         //env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
  17.         //env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);
  18.  
  19.         //env.setRestartStrategy(RestartStrategies.fixedDelayRestart());
  20.  
  21.         //TODO 2.消费 ods_base_log 主题数据创建流
  22.         String sourceTopic = "ods_base_log";
  23.         String groupId = "base_log_app_210325";
  24.         DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(sourceTopic, groupId));
  25.  
  26.         //TODO 3.将每行数据转换为JSON对象
  27.         OutputTag<String> outputTag = new OutputTag<String>("Dirty") {
  28.         };
  29.         SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.process(new ProcessFunction<String, JSONObject>() {
  30.             @Override
  31.             public void processElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
  32.                 try {
  33.                     JSONObject jsonObject = JSON.parseObject(value);
  34.                     out.collect(jsonObject);
  35.                 } catch (Exception e) {
  36.                     //发生异常,将数据写入侧输出流
  37.                     ctx.output(outputTag, value);
  38.                 }
  39.             }
  40.         });
  41.  
  42.         //打印脏数据
  43.         jsonObjDS.getSideOutput(outputTag).print("Dirty>>>>>>>>>>>");
  44.  
  45.         //TODO 4.新老用户校验  状态编程
  46.         SingleOutputStreamOperator<JSONObject> jsonObjWithNewFlagDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"))
  47.                 .map(new RichMapFunction<JSONObject, JSONObject>() {
  48.  
  49.                     private ValueState<String> valueState;
  50.  
  51.                     @Override
  52.                     public void open(Configuration parameters) throws Exception {
  53.                         valueState = getRuntimeContext().getState(new ValueStateDescriptor<String>("value-state", String.class));
  54.                     }
  55.  
  56.                     @Override
  57.                     public JSONObject map(JSONObject value) throws Exception {
  58.  
  59.                         //获取数据中的"is_new"标记
  60.                         String isNew = value.getJSONObject("common").getString("is_new");
  61.  
  62.                         //判断isNew标记是否为"1"
  63.                         if ("1".equals(isNew)) {
  64.  
  65.                             //获取状态数据
  66.                             String state = valueState.value();
  67.  
  68.                             if (state != null) {
  69.                                 //修改isNew标记
  70.                                 value.getJSONObject("common").put("is_new", "0");
  71.                             } else {
  72.                                 valueState.update("1");
  73.                             }
  74.                         }
  75.  
  76.                         return value;
  77.                     }
  78.                 });
  79.  
  80.         //TODO 5.分流  侧输出流  页面：主流  启动：侧输出流  曝光：侧输出流
  81.         OutputTag<String> startTag = new OutputTag<String>("start") {
  82.         };
  83.         OutputTag<String> displayTag = new OutputTag<String>("display") {
  84.         };
  85.         SingleOutputStreamOperator<String> pageDS = jsonObjWithNewFlagDS.process(new ProcessFunction<JSONObject, String>() {
  86.             @Override
  87.             public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {
  88.  
  89.                 //获取启动日志字段
  90.                 String start = value.getString("start");
  91.                 if (start != null && start.length() > 0) {
  92.                     //将数据写入启动日志侧输出流
  93.                     ctx.output(startTag, value.toJSONString());
  94.                 } else {
  95.                     //将数据写入页面日志主流
  96.                     out.collect(value.toJSONString());
  97.  
  98.                     //取出数据中的曝光数据
  99.                     JSONArray displays = value.getJSONArray("displays");
  100.  
  101.                     if (displays != null && displays.size() > 0) {
  102.  
  103.                         //获取页面ID
  104.                         String pageId = value.getJSONObject("page").getString("page_id");
  105.  
  106.                         for (int i = 0; i < displays.size(); i++) {
  107.                             JSONObject display = displays.getJSONObject(i);
  108.  
  109.                             //添加页面id
  110.                             display.put("page_id", pageId);
  111.  
  112.                             //将输出写出到曝光侧输出流
  113.                             ctx.output(displayTag, display.toJSONString());
  114.                         }
  115.                     }
  116.                 }
  117.             }
  118.         });
  119.  
  120.         //TODO 6.提取侧输出流
  121.         DataStream<String> startDS = pageDS.getSideOutput(startTag);
  122.         DataStream<String> displayDS = pageDS.getSideOutput(displayTag);
  123.  
  124.         //TODO 7.将三个流进行打印并输出到对应的Kafka主题中
  125.         startDS.print("Start>>>>>>>>>>>");
  126.         pageDS.print("Page>>>>>>>>>>>");
  127.         displayDS.print("Display>>>>>>>>>>>>");
  128.  
  129.         startDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_start_log"));
  130.         pageDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_page_log"));
  131.         displayDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_display_log"));
  132.  
  133.         //TODO 8.启动任务
  134.         env.execute("BaseLogApp");
  135.  
  136.     }
  137.  
  138. }

 

 

054- DWD&DIM-业务数据只代码编写 开启配置表Binlog并测试

开启Binlog：

  1. 修改my.conf文件：/etc
  2. ##监控多个库
  3. [mysqld]
  4. # log_bin
  5. log-bin = mysql-bin 
  6. binlog-format = ROW 
  7. server_id = 1 
  8. binlog-do-db=gmall-210325-flink
  9. binlog-do-db=gmall-210325-realtime

063-DWD&DIM-业务数据之代码编写测试

067-DWD&DIM-业务数据之整体测试  测试完成

  1. 启动zookeeper:
  2.    bin/zkServer.sh start 
  3. 启动hdfs:
  4.    sbin/hadoop-daemon.sh start namenode
  5.    sbin/hadoop-daemon.sh start datanode
  6. 启动hbase:
  7.    bin/hbase-daemon.sh start master
  8.    bin/hbase-daemon.sh start regionserver
  9. 启动kafaka:
  10.    启动kafka:
  11.    bin/kafka-server-start.sh config/server9092.properties 
  12.    启动消费者"ods_base_db":
  13.    bin/kafka-console-consumer.sh --bootstrap-server  bigdata- 
  14.    training01.erongda.com:9092 -from-beginning --topic  ods_base_db
  15.  
  16. 启动程序：
  17.    BaseDBApp.java 
  18.    ods/FlinkCDC.java
  19.  
  20. 数据流：web/app -> nginx -> SpringBoot -> Mysql -> FlinkApp -> Kafka(ods) -> FlinkApp -> Kafka(dwd)/Phoenix(dim)
  21. 程  序：                        mockDb -> Mysql -> FlinkCDC -> Kafka(ZK) -> BaseDBApp -> Kafka/Phoenix(hbase,zk,hdfs)
  22.  
  23.  
  24.
注意：为了开启 hbase 的 namespace 和 phoenix 的 schema 的映射，在程序中需要加这
个配置文件，另外在 linux 服务上，也需要在 hbase 以及 phoenix 的 hbase-site.xml 配置

文件中，加上以上两个配置，并使用 xsync 进行同步。

  1. <property>
  2.      <name>phoenix.schema.isNamespaceMappingEnabled</name>
  3.      <value>true</value>
  4. </property>
  5. <property>
  6.      <name>phoenix.schema.mapSystemTablesToNamespace</name>
  7.      <value>true</value>
  8. </property>

 

 启动phoenix后新建schema:

create  schema  GMALL210_REALTIME;

测试：

1.phoenix数据表启动时候或者修改后是否新建

    1.1 启动BaseDBApp新建表： 

 

    1.2  新增、修改、删除gmall-210325-flink.table_process也会去创建表

2.type为hbase数据是否正确输出

  

3.type为kafka的数据是否正确输出

bin/kafka-console-consumer.sh --bootstrap-server  bigdata-training01.erongda.com:9092 -from-beginning --topic dwd_order_info 

078-DWM层-访客UV  代码测试

  1. 启动zookeeper:
  2.    bin/zkServer.sh start 
  3.  
  4. 启动kafaka:
  5.    启动kafka:
  6.    bin/kafka-server-start.sh config/server9092.properties 
  7.    
  8.    启动生产者：
  9.    bin/kafka-console-producer.sh --broker-list  bigdata-training01.erongda.com:9092 
  10.    --topic dwd_page_log
  11.  
  12.    启动消费者:
  13.    bin/kafka-console-consumer.sh --bootstrap-server  bigdata- 
  14.    training01.erongda.com:9092 -from-beginning --topic  dwm_unique_visit
  15.  
  16. 启动：
  17.    java -jar gma112020-mock-1og-2020-12-18.iar
  18.    java -jar gmall-logger.jar 
  19.  
  20. 启动：
  21.   BaseLogApp.java 
  22.   UniqueVisitApp.java 
  23.  
  24. 测试数据：
  25. 去除last_page_id：
  26. {"common":{"ar":"310000","ba":"Redmi","ch":"xiaomi","is_new":"1","md":"Redmi k30","mid":"mid7","os":"Android 11.0","uid":"23","vc":"v2.1.111"},"page":{"during_time":10393,"item":"1,9","item_type":"sku_ids","page_id":"trade"},"ts":1690869978000}
  27.  
  28. 换一个mid：
  29. {"common":{"ar":"310000","ba":"Redmi","ch":"xiaomi","is_new":"1","md":"Redmi k30","mid":"mid_8","os":"Android 11.0","uid":"23","vc":"v2.1.111"},"page":{"during_time":10393,"item":"1,9","item_type":"sku_ids","page_id":"trade"},"ts":1690869978000}
  30.  
  31. //数据流：web/app -> Nginx -> SpringBoot -> Kafka(ods) -> FlinkApp -> Kafka(dwd)
  32. //程  序：mockLog -> Nginx -> Logger.sh  -> Kafka(ZK)  -> BaseLogApp -> kafka
  33.

 

自测：

 造行为数据测试：   java -jar gma112020-mock-1og-2020-12-18.iar

081-DWM层-跳出明细  代码测试

  1. 启动zookeeper:
  2.    bin/zkServer.sh start 
  3.  
  4. 启动kafaka:
  5.    启动kafka:
  6.    bin/kafka-server-start.sh config/server9092.properties 
  7.    
  8.    启动生产者：
  9.    bin/kafka-console-producer.sh --broker-list  bigdata-training01.erongda.com:9092 
  10.    --topic dwd_page_log
  11.  
  12.    启动消费者:
  13.    bin/kafka-console-consumer.sh --bootstrap-server  bigdata-training01.erongda.com:9092 -from-beginning --topic  dwm_user_jump_detail
  14.  
  15. 启动：
  16.    UserJumpDetailApp.java 
  17.  
  18. 测试数据：
  19. 去除last_page_id：
  20. {"common":{"ar":"310000","ba":"Redmi","ch":"xiaomi","is_new":"1","md":"Redmi k30","mid":"mid_8","os":"Android 11.0","uid":"23","vc":"v2.1.111"},"page":{"during_time":10393,"item":"1,9","item_type":"sku_ids","page_id":"trade"},"ts":1690889970000}
  21.  
  22. {"common":{"ar":"310000","ba":"Redmi","ch":"xiaomi","is_new":"1","md":"Redmi k30","mid":"mid_8","os":"Android 11.0","uid":"23","vc":"v2.1.111"},"page":{"during_time":10393,"item":"1,9","item_type":"sku_ids","page_id":"trade"},"ts":1690889975000}
  23.  
  24. {"common":{"ar":"310000","ba":"Redmi","ch":"xiaomi","is_new":"1","md":"Redmi k30","mid":"mid_8","os":"Android 11.0","uid":"23","vc":"v2.1.111"},"page":{"during_time":10393,"item":"1,9","item_type":"sku_ids","page_id":"trade"},"ts":1690889979000}
  25.  
  26. {"common":{"ar":"310000","ba":"Redmi","ch":"xiaomi","is_new":"1","md":"Redmi k30","mid":"mid_8","os":"Android 11.0","uid":"23","vc":"v2.1.111"},"page":{"during_time":10393,"item":"1,9","item_type":"sku_ids","page_id":"trade"},"ts":1690889992000}
  27.  
  28. //数据流：web/app -> Nginx -> SpringBoot -> Kafka(ods) -> FlinkApp -> Kafka(dwd) -> FlinkApp -> Kafka(dwm)
  29. //程  序：mockLog -> Nginx -> Logger.sh  -> Kafka(ZK)  -> BaseLogApp -> kafka -> UserJumpDetailApp -> Kafka

 

082-DWM层-跳出明细  测试

  1. 启动zookeeper:
  2.    bin/zkServer.sh start 
  3.  
  4. 启动kafaka:
  5.    启动kafka:
  6.    bin/kafka-server-start.sh config/server9092.properties 
  7.    
  8.    启动消费者:
  9.    bin/kafka-console-consumer.sh --bootstrap-server  bigdata-training01.erongda.com:9092 -from-beginning --topic  dwm_user_jump_detail
  10.  
  11. 启动：
  12.    java -jar gma112020-mock-1og-2020-12-18.iar
  13.    java -jar gmall-logger.jar 
  14.  
  15. 启动：
  16.   BaseLogApp.java 
  17.   UserJumpDetailApp.java
  18.

091-DWM层-订单宽表  代码测试  测试完成

  1. 启动zookeeper:
  2.    bin/zkServer.sh start 
  3. 启动hdfs:
  4.    sbin/hadoop-daemon.sh start namenode
  5.    sbin/hadoop-daemon.sh start datanode
  6. 启动hbase:
  7.    bin/hbase-daemon.sh start master
  8.    bin/hbase-daemon.sh start regionserver
  9. 启动kafaka:
  10.    启动kafka:
  11.    bin/kafka-server-start.sh config/server9092.properties 
  12.    启动消费者"ods_base_db":
  13.    bin/kafka-console-consumer.sh --bootstrap-server  bigdata- 
  14.    training01.erongda.com:9092 -from-beginning --topic  ods_base_db
  15.  
  16.    bin/kafka-console-consumer.sh --bootstrap-server  bigdata- 
  17.    training01.erongda.com:9092 -from-beginning --topic  dwd_order_detail
  18.  
  19.    bin/kafka-console-consumer.sh --bootstrap-server  bigdata- 
  20.    training01.erongda.com:9092 -from-beginning --topic  dwd_order_info
  21.  
  22.    bin/kafka-console-consumer.sh --bootstrap-server  bigdata- 
  23.    training01.erongda.com:9092 -from-beginning --topic  dwm_order_wide
  24.  
  25. 启动程序：
  26.    BaseDBApp.java 
  27.    ods/FlinkCDC.java
  28.    OrderWideApp.java
  29.  
  30.  
  31. 注意：mock.clear=1 每次都会置空数据库,便于测试
  32.  
  33. 数据流：web/app -> nginx -> SpringBoot -> Mysql -> FlinkApp -> Kafka(ods) -> FlinkApp -> Kafka/Phoenix(dwd-dim) -> FlinkApp(redis) -> Kafka(dwm)
  34. 程  序：         MockDb               -> Mysql -> FlinkCDC -> Kafka(ZK) -> BaseDbApp -> Kafka/Phoenix(zk/hdfs/hbase) -> OrderWideApp(Redis) -> Kafka

测试：

1.手动创建订单数据

2.测试数据有无丢失

3.相关数据的实例

BaseDBApp:

Kafka>>>>>>>>:2> {"sinkTable":"dwd_order_detail","database":"gmall-210325-flink","before":{},"after":{"sku_num":"2","create_time":"2023-06-12 16:33:42","sku_id":20,"order_price":2899.00,"source_type":"2401","sku_name":"小米电视E65X 65英寸 全面屏 4K超高清HDR 蓝牙遥控内置小爱 2+8GB AI人工智能液晶网络平板电视 L65M5-EA","id":79949,"order_id":26689,"split_total_amount":5798.00},"type":"insert","tableName":"order_detail"}

FlinkCDC:

{"database":"gmall-210325-flink","before":{},"after":{"sku_num":"2","create_time":"2023-06-12 16:33:42","sku_id":20,"order_price":2899.00,"source_type":"2401","img_url":

"http://47.93.148.192:8080/group1/M00/00/02/rBHu8l-0kIGAWtMyAAGxs6Q350k510.jpg","sku_name":"小米电视E65X 65英寸 全面屏 4K超高清HDR 蓝牙遥控内置小爱 2+8GB AI人工智能液晶网络平板电视 L65M5-EA","id":79949,"order_id":26689,"split_total_amount":5798.00},"type":"insert","tableName":"order_detail"}

OrderWideApp:
orderWideWithNoDimDS>>>>>>>>>> OrderWide(detail_id=79949, order_id=26689, sku_id=20, order_price=2899.00, sku_num=2, sku_name=小米电视E65X 65英寸 全面屏 4K超高清HDR 蓝牙遥控内置小爱 2+8GB AI人工智能液晶网络平板电视 L65M5-EA, province_id=19, order_status=1001, user_id=316, total_amount=10490.00, activity_reduce_amount=0.00, coupon_reduce_amount=0.00, original_total_amount=10484.00, feight_fee=6.00, split_feight_fee=null, split_activity_amount=null, split_coupon_amount=null, split_total_amount=5798.00, expire_time=null, create_time=2023-06-12 16:33:42, operate_time=null, create_date=2023-06-12, create_hour=16, province_name=null, province_area_code=null, province_iso_code=null, province_3166_2_code=null, user_age=null, user_gender=null, spu_id=null, tm_id=null, category3_id=null, spu_name=null, tm_name=null, category3_name=null)

orderWideWithCategory3DS>>>>>>>>>>>> OrderWide(detail_id=79949, order_id=26689, sku_id=20, order_price=2899.00, sku_num=2, sku_name=小米电视E65X 65英寸 全面屏 4K超高清HDR 蓝牙遥控内置小爱 2+8GB AI人工智能液晶网络平板电视 L65M5-EA, province_id=19, order_status=1001, user_id=316, total_amount=10490.00, activity_reduce_amount=0.00, coupon_reduce_amount=0.00, original_total_amount=10484.00, feight_fee=6.00, split_feight_fee=null, split_activity_amount=null, split_coupon_amount=null, split_total_amount=5798.00, expire_time=null, create_time=2023-06-12 16:33:42, operate_time=null, create_date=2023-06-12, create_hour=16, province_name=甘肃, province_area_code=620000, province_iso_code=CN-62, province_3166_2_code=CN-GS, user_age=50, user_gender=F, spu_id=6, tm_id=5, category3_id=86, spu_name=小米电视 内置小爱 智能网络液晶平板教育电视, tm_name=小米, category3_name=平板电视)

095-DWM层-订单宽表  关联维度 JDBCUtil 测试

  1.     启动JdbcUtil
  2.     public static void main(String[] args) throws Exception {
  3.  
  4.         Class.forName(GmallConfig.PHOENIX_DRIVER);
  5.         Connection connection = 
  6.         DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
  7.  
  8.         List<JSONObject> queryList = queryList(connection,
  9.                 "select * from GMALL210325_REALTIME.DIM_USER_INFO",
  10.                 JSONObject.class,
  11.                 true);
  12.  
  13.         for (JSONObject jsonObject : queryList) {
  14.             System.out.println(jsonObject);
  15.         }
  16.  
  17.         connection.close();
  18.  
  19.     }
测试mysql、clickhouse和日志数据是否一致：

 100--DWM层-订单宽表  关联维度  优化1旁路缓存  代码测试

  1. package com.atguigu.utils;
  2.  
  3. import com.alibaba.fastjson.JSONObject;
  4. import com.atguigu.common.GmallConfig;
  5. import redis.clients.jedis.Jedis;
  6.  
  7. import java.sql.Connection;
  8. import java.sql.DriverManager;
  9. import java.util.List;
  10.  
  11. public class DimUtil {
  12.  
  13.     public static JSONObject getDimInfo(Connection connection, String tableName, String id) throws Exception {
  14.  
  15.         //查询Phoenix之前先查询Redis
  16.         Jedis jedis = RedisUtil.getJedis();
  17.         //DIM:DIM_USER_INFO:143
  18.         String redisKey = "DIM:" + tableName + ":" + id;
  19.         String dimInfoJsonStr = jedis.get(redisKey);
  20.         if (dimInfoJsonStr != null) {
  21.             //重置过期时间
  22.             jedis.expire(redisKey, 24 * 60 * 60);
  23.             //归还连接
  24.             jedis.close();
  25.             //返回结果
  26.             return JSONObject.parseObject(dimInfoJsonStr);
  27.         }
  28.  
  29.         //拼接查询语句
  30.         //select * from db.tn where id='18';
  31.         String querySql = "select * from " + GmallConfig.HBASE_SCHEMA + "." + tableName +
  32.                 " where id='" + id + "'";
  33.  
  34.         //查询Phoenix
  35.         List<JSONObject> queryList = JdbcUtil.queryList(connection, querySql, JSONObject.class, false);
  36.         JSONObject dimInfoJson = queryList.get(0);
  37.  
  38.         //在返回结果之前,将数据写入Redis
  39.         jedis.set(redisKey, dimInfoJson.toJSONString());
  40.         jedis.expire(redisKey, 24 * 60 * 60);
  41.         jedis.close();
  42.  
  43.         //返回结果
  44.         return dimInfoJson;
  45.     }
  46.  
  47.     public static void delRedisDimInfo(String tableName, String id) {
  48.         Jedis jedis = RedisUtil.getJedis();
  49.         String redisKey = "DIM:" + tableName + ":" + id;
  50.         jedis.del(redisKey);
  51.         jedis.close();
  52.     }
  53.  
  54.     public static void main(String[] args) throws Exception {
  55.  
  56.         Class.forName(GmallConfig.PHOENIX_DRIVER);
  57.         Connection connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
  58.  
  59.         long start = System.currentTimeMillis();
  60.         //System.out.println(getDimInfo(connection, "DIM_BASE_TRADEMARK", "15 "));
  61.         System.out.println(getDimInfo(connection, "DIM_USER_INFO", "143"));
  62.         long end = System.currentTimeMillis();
  63.         System.out.println(getDimInfo(connection, "DIM_USER_INFO", "143"));
  64.         long end2 = System.currentTimeMillis();
  65.         System.out.println(getDimInfo(connection, "DIM_USER_INFO", "143"));
  66.         long end3 = System.currentTimeMillis();
  67.  
  68.       System.out.println(end - start);
  69.       System.out.println(end2 - end);
  70.       System.out.println(end3 - end2);
  71.  
  72.         connection.close();
  73.  
  74.     }
  75.  
  76. }

107-DWM层-订单宽表  关联维度  优化2  异步IO编码  测试完成

  1. 启动zookeeper:
  2.    bin/zkServer.sh start 
  3.  
  4. 启动hdfs:
  5.    sbin/hadoop-daemon.sh start namenode
  6.    sbin/hadoop-daemon.sh start datanode
  7.  
  8. 启动hbase:
  9.    bin/hbase-daemon.sh start master
  10.    bin/hbase-daemon.sh start regionserver
  11.  
  12. 启动程序：
  13.    BaseDBApp.java 
  14.    ods/FlinkCDC.java
  15.    OrderWideApp.java
  16.  
  17. 启动：
  18.    gmall2020-mock-db-2020-11-27.jar
  19.  
  20. select  count(*) from  `gmall-210325-flink`.order_detail

 

109-DWM层-订单宽表  最终测试

bin/kafka-console-consumer.sh --bootstrap-server  bigdata-training01.erongda.com:9092 -from-beginning --topic  dwm_order_wide 

113-尚硅谷-Flink实时数仓-DWM层-支付宽表  代码测试

  1. 启动kafaka:
  2.    bin/kafka-console-consumer.sh --bootstrap-server  bigdata- 
  3.    training01.erongda.com:9092 -from-beginning --topic  dwm_payment_wide
  4.  
  5. 启动程序：
  6.    BaseDBApp.java 
  7.    ods/FlinkCDC.java
  8.    OrderWideApp.java
  9.    PaymentWideApp.java 
  10.  
  11.  SELECT
  12. 	count(*) 
  13.  FROM
  14. 	payment_info p
  15. 	JOIN order_detail o ON p.order_id = o.order_id;

 测试数据是否一致：

 

 相关数据的例子：

  1. 注意：对应数据关联上了，对应的维度数据也查询到了
  2. OrderWideApp：
  3. orderWideWithCategory3DS>>>>>>>>>>>> OrderWide(detail_id=80399, order_id=26873, sku_id=16, order_price=4488.00, sku_num=1, sku_name=华为 HUAWEI P40 麒麟990 5G SoC芯片 5000万超感知徕卡三摄 30倍数字变焦 8GB+128GB亮黑色全网通5G手机, province_id=16, order_status=1001, user_id=3209, total_amount=13959.00, activity_reduce_amount=0.00, coupon_reduce_amount=0.00, original_total_amount=13945.00, feight_fee=14.00, split_feight_fee=null, split_activity_amount=null, split_coupon_amount=null, split_total_amount=4488.00, expire_time=null, create_time=2023-06-12 19:59:53, operate_time=null, create_date=2023-06-12, create_hour=19, province_name=吉林, province_area_code=220000, province_iso_code=CN-22, province_3166_2_code=CN-JL, user_age=18, user_gender=F, spu_id=4, tm_id=3, category3_id=61, spu_name=HUAWEI P40, tm_name=华为, category3_name=手机)
  4.  
  5. orderWideWithCategory3DS>>>>>>>>>>>> OrderWide(detail_id=80404, order_id=26874, sku_id=31, order_price=69.00, sku_num=1, sku_name=CAREMiLLE珂曼奶油小方口红 雾面滋润保湿持久丝缎唇膏 M03赤茶, province_id=32, order_status=1001, user_id=3539, total_amount=20366.00, activity_reduce_amount=0.00, coupon_reduce_amount=0.00, original_total_amount=20361.00, feight_fee=5.00, split_feight_fee=null, split_activity_amount=null, split_coupon_amount=null, split_total_amount=69.00, expire_time=null, create_time=2023-06-12 19:59:53, operate_time=null, create_date=2023-06-12, create_hour=19, province_name=贵州, province_area_code=520000, province_iso_code=CN-52, province_3166_2_code=CN-GZ, user_age=24, user_gender=M, spu_id=10, tm_id=9, category3_id=477, spu_name=CAREMiLLE珂曼奶油小方口红 雾面滋润保湿持久丝缎唇膏, tm_name=CAREMiLLE, category3_name=唇部)
  6.  
  7. PaymentWideApp：
  8. >>>>>>>>>> PaymentWide(payment_id=18002, subject=华为 HUAWEI P40 麒麟990 5G SoC芯片 5000万超感知徕卡三摄 30倍数字变焦 8GB+128GB亮黑色全网通5G手机等4件商品, payment_type=1101, payment_create_time=2023-06-12 19:59:53, callback_time=2023-06-12 20:00:13, detail_id=80399, order_id=26873, sku_id=16, order_price=4488.00, sku_num=1, sku_name=华为 HUAWEI P40 麒麟990 5G SoC芯片 5000万超感知徕卡三摄 30倍数字变焦 8GB+128GB亮黑色全网通5G手机, province_id=16, order_status=1001, user_id=3209, total_amount=13959.00, activity_reduce_amount=0.00, coupon_reduce_amount=0.00, original_total_amount=13945.00, feight_fee=14.00, split_feight_fee=null, split_activity_amount=null, split_coupon_amount=null, split_total_amount=4488.00, order_create_time=2023-06-12 19:59:53, province_name=吉林, province_area_code=220000, province_iso_code=CN-22, province_3166_2_code=CN-JL, user_age=18, user_gender=F, spu_id=4, tm_id=3, category3_id=61, spu_name=HUAWEI P40, tm_name=华为, category3_name=手机)
  9.  
  10. >>>>>>>>>> PaymentWide(payment_id=17996, subject=TCL 85Q6 85英寸 巨幕私人影院电视 4K超高清 AI智慧屏 全景全面屏 MEMC运动防抖 2+16GB 液晶平板电视机等8件商品, payment_type=1102, payment_create_time=2023-06-12 19:59:53, callback_time=2023-06-12 20:00:13, detail_id=80382, order_id=26864, sku_id=16, order_price=4488.00, sku_num=2, sku_name=华为 HUAWEI P40 麒麟990 5G SoC芯片 5000万超感知徕卡三摄 30倍数字变焦 8GB+128GB亮黑色全网通5G手机, province_id=8, order_status=1001, user_id=1008, total_amount=45106.00, activity_reduce_amount=0.00, coupon_reduce_amount=0.00, original_total_amount=45090.00, feight_fee=16.00, split_feight_fee=null, split_activity_amount=null, split_coupon_amount=null, split_total_amount=8976.00, order_create_time=2023-06-12 19:59:53, province_name=浙江, province_area_code=330000, province_iso_code=CN-33, province_3166_2_code=CN-ZJ, user_age=56, user_gender=F, spu_id=4, tm_id=3, category3_id=61, spu_name=HUAWEI P40, tm_name=华为, category3_name=手机)

120-Flink实时数仓-DWS层-访客主题宽表  打印测试

  1. 启动:
  2.     BaseLogApp.java 
  3.     UniqueVisitApp.java 
  4.     UserJumpDetailApp.java 
  5.     VisitorStatsApp.java 
  6. 启动日志:
  7.    java -jar gmall2020-mock-log-2020-12-18.jar
  8.    java -jar gmall-logger.jar

 

  1. UserJumpDetailApp:
  2. {"common":{"ar":"310000","uid":"1","os":"iOS 13.2.3","ch":"Appstore","is_new":"0","md":"iPhone 8","mid":"mid_20","vc":"v2.1.134","ba":"iPhone"},"page":{"page_id":"home","during_time":2584},"displays":[{"display_type":"activity","item":"1","item_type":"activity_id","pos_id":4,"order":1},{"display_type":"activity","item":"1","item_type":"activity_id","pos_id":4,"order":2},{"display_type":"query","item":"9","item_type":"sku_id","pos_id":2,"order":3},{"display_type":"query","item":"7","item_type":"sku_id","pos_id":5,"order":4},{"display_type":"promotion","item":"3","item_type":"sku_id","pos_id":1,"order":5},{"display_type":"query","item":"10","item_type":"sku_id","pos_id":1,"order":6},{"display_type":"query","item":"8","item_type":"sku_id","pos_id":1,"order":7},{"display_type":"query","item":"4","item_type":"sku_id","pos_id":4,"order":8}],"ts":1690880391000}
  3. {"common":{"ar":"110000","uid":"31","os":"Android 11.0","ch":"huawei","is_new":"0","md":"Xiaomi 10 Pro ","mid":"mid_20","vc":"v2.1.134","ba":"Xiaomi"},"page":{"page_id":"home","during_time":19457},"displays":[{"display_type":"activity","item":"1","item_type":"activity_id","pos_id":5,"order":1},{"display_type":"activity","item":"1","item_type":"activity_id","pos_id":5,"order":2},{"display_type":"promotion","item":"3","item_type":"sku_id","pos_id":2,"order":3},{"display_type":"query","item":"9","item_type":"sku_id","pos_id":1,"order":4},{"display_type":"recommend","item":"10","item_type":"sku_id","pos_id":2,"order":5},{"display_type":"query","item":"5","item_type":"sku_id","pos_id":4,"order":6},{"display_type":"query","item":"1","item_type":"sku_id","pos_id":2,"order":7},{"display_type":"promotion","item":"2","item_type":"sku_id","pos_id":3,"order":8},{"display_type":"promotion","item":"8","item_type":"sku_id","pos_id":5,"order":9},{"display_type":"query","item":"5","item_type":"sku_id","pos_id":4,"order":10},{"display_type":"promotion","item":"7","item_type":"sku_id","pos_id":4,"order":11}],"ts":1690880394000}
  4.  
  5. VisitorStatsApp:
  6. >>>>>>>>>>>> VisitorStats(stt=2023-08-01 16:59:20, edt=2023-08-01 16:59:30, vc=v2.0.1, ch=Appstore, ar=310000, is_new=0, uv_ct=1, pv_ct=9, sv_ct=1, uj_ct=0, dur_sum=116391, ts=1690880362000)
  7. >>>>>>>>>>>> VisitorStats(stt=2023-08-01 16:59:20, edt=2023-08-01 16:59:30, vc=v2.1.132, ch=xiaomi, ar=110000, is_new=0, uv_ct=1, pv_ct=2, sv_ct=1, uj_ct=0, dur_sum=16767, ts=1690880364000)
  8. >>>>>>>>>>>> VisitorStats(stt=2023-08-01 16:59:20, edt=2023-08-01 16:59:30, vc=v2.1.134, ch=oppo, ar=310000, is_new=0, uv_ct=1, pv_ct=9, sv_ct=1, uj_ct=0, dur_sum=111592, ts=1690880365000)
  9. >>>>>>>>>>>> VisitorStats(stt=2023-08-01 16:59:20, edt=2023-08-01 16:59:30, vc=v2.1.134, ch=Appstore, ar=420000, is_new=0, uv_ct=0, pv_ct=2, sv_ct=1, uj_ct=0, dur_sum=29586, ts=1690880369000)
  10. >>>>>>>>>>>> VisitorStats(stt=2023-08-01 16:59:20, edt=2023-08-01 16:59:30, vc=v2.1.132, ch=Appstore, ar=420000, is_new=0, uv_ct=1, pv_ct=2, sv_ct=1, uj_ct=0, dur_sum=8265, ts=1690880369000)

uj_ct都为0的原因：

 解决方案：
     方案1：将事件改为处理时间：但这样不太好，消费相同的数据不具备幂等性
     方案2：水准线改为11秒 需要损失一定的时效性

  1. VisitorStatsApp.java 
  2. //TODO 5.提取时间戳生成WaterMark
  3. SingleOutputStreamOperator<VisitorStats> visitorStatsWithWMDS = unionDS.assignTimestampsAndWatermarks(WatermarkStrategy
  4. .<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(11))
  5. .withTimestampAssigner(new SerializableTimestampAssigner<VisitorStats>() {
  6.     @Override
  7.     public long extractTimestamp(VisitorStats element, long recordTimestamp) {
  8.         return element.getTs();
  9.     }
  10. }));

143-DWS层-访客主题  ClickHouseUtil  测试完成

  1. 启动clickhouse:
  2.    docker exec -it clickhouse-server /bin/bash
  3.    clickhouse-client
  4.  
  5. 创建clickhouse表:
  6. create table visitor_stats_210325 (
  7.  stt DateTime,
  8.  edt DateTime,
  9.  vc String,
  10.  ch String,
  11.  ar String,
  12.  is_new String,
  13.  uv_ct UInt64,
  14.  pv_ct UInt64,
  15.  sv_ct UInt64,
  16.  uj_ct UInt64,
  17.  dur_sum UInt64,
  18.  ts UInt64
  19.  ) engine =ReplacingMergeTree(ts)
  20.  partition by toYYYYMMDD(stt)
  21.  order by (stt,edt,is_new,vc,ch,ar);
  22.  
  23. 启动程序:
  24.    BaseLogApp.java 
  25.    UniqueVisitApp.java
  26.    UserJumpDetailApp.java
  27.    VisitorStatsApp.java  
  28.  
  29.

造日志数据： /opt/modules/gmall-flink/rt_applog

 sudo java -jar gmall-logger.jar 

 sudo java -jar gmall2020-mock-log-2020-12-18.jar 

 select count(*) from visitor_stats_210325;

注意：插入是5的倍数，所以为55条

 

  1. 相关数据：
  2. UniqueVisitApp：
  3. {"common":{"ar":"310000","uid":"42","os":"Android 10.0","ch":"web","is_new":"0","md":"Oneplus 7","mid":"mid_6","vc":"v2.1.134","ba":"Oneplus"},"page":{"page_id":"home","during_time":14678},"displays":[{"display_type":"activity","item":"2","item_type":"activity_id","pos_id":5,"order":1},{"display_type":"activity","item":"1","item_type":"activity_id","pos_id":5,"order":2},{"display_type":"query","item":"3","item_type":"sku_id","pos_id":1,"order":3},{"display_type":"query","item":"1","item_type":"sku_id","pos_id":4,"order":4},{"display_type":"query","item":"1","item_type":"sku_id","pos_id":4,"order":5},{"display_type":"recommend","item":"9","item_type":"sku_id","pos_id":3,"order":6},{"display_type":"query","item":"9","item_type":"sku_id","pos_id":3,"order":7}],"ts":1690880847000}
  4.  
  5. {"common":{"ar":"110000","uid":"5","os":"Android 11.0","ch":"oppo","is_new":"0","md":"Sumsung Galaxy S20","mid":"mid_15","vc":"v2.1.134","ba":"Sumsung"},"page":{"page_id":"home","during_time":2184},"displays":[{"display_type":"activity","item":"1","item_type":"activity_id","pos_id":1,"order":1},{"display_type":"activity","item":"1","item_type":"activity_id","pos_id":1,"order":2},{"display_type":"promotion","item":"6","item_type":"sku_id","pos_id":5,"order":3},{"display_type":"promotion","item":"9","item_type":"sku_id","pos_id":1,"order":4},{"display_type":"query","item":"7","item_type":"sku_id","pos_id":2,"order":5},{"display_type":"promotion","item":"2","item_type":"sku_id","pos_id":5,"order":6},{"display_type":"promotion","item":"5","item_type":"sku_id","pos_id":5,"order":7},{"display_type":"recommend","item":"2","item_type":"sku_id","pos_id":5,"order":8},{"display_type":"query","item":"10","item_type":"sku_id","pos_id":4,"order":9},{"display_type":"promotion","item":"9","item_type":"sku_id","pos_id":3,"order":10},{"display_type":"query","item":"3","item_type":"sku_id","pos_id":1,"order":11},{"display_type":"query","item":"8","item_type":"sku_id","pos_id":1,"order":12}],"ts":1690880874000}
  6.  
  7. UserJumpDetailApp：
  8. {"common":{"ar":"370000","uid":"20","os":"Android 11.0","ch":"web","is_new":"0","md":"Xiaomi Mix2 ","mid":"mid_16","vc":"v2.1.134","ba":"Xiaomi"},"page":{"page_id":"home","during_time":9663},"displays":[{"display_type":"activity","item":"2","item_type":"activity_id","pos_id":4,"order":1},{"display_type":"recommend","item":"10","item_type":"sku_id","pos_id":2,"order":2},{"display_type":"promotion","item":"7","item_type":"sku_id","pos_id":3,"order":3},{"display_type":"query","item":"3","item_type":"sku_id","pos_id":5,"order":4},{"display_type":"query","item":"3","item_type":"sku_id","pos_id":1,"order":5},{"display_type":"query","item":"10","item_type":"sku_id","pos_id":5,"order":6},{"display_type":"recommend","item":"8","item_type":"sku_id","pos_id":2,"order":7}],"ts":1690880883000}
  9.  
  10. {"common":{"ar":"110000","uid":"1","os":"iOS 13.3.1","ch":"Appstore","is_new":"0","md":"iPhone 8","mid":"mid_9","vc":"v2.1.111","ba":"iPhone"},"page":{"page_id":"home","during_time":16573},"displays":[{"display_type":"activity","item":"2","item_type":"activity_id","pos_id":4,"order":1},{"display_type":"query","item":"9","item_type":"sku_id","pos_id":4,"order":2},{"display_type":"promotion","item":"7","item_type":"sku_id","pos_id":1,"order":3},{"display_type":"promotion","item":"8","item_type":"sku_id","pos_id":3,"order":4},{"display_type":"promotion","item":"8","item_type":"sku_id","pos_id":5,"order":5},{"display_type":"query","item":"3","item_type":"sku_id","pos_id":3,"order":6},{"display_type":"query","item":"1","item_type":"sku_id","pos_id":1,"order":7},{"display_type":"query","item":"8","item_type":"sku_id","pos_id":3,"order":8},{"display_type":"recommend","item":"6","item_type":"sku_id","pos_id":2,"order":9},{"display_type":"query","item":"1","item_type":"sku_id","pos_id":1,"order":10},{"display_type":"query","item":"9","item_type":"sku_id","pos_id":4,"order":11}],"ts":1690880894000}
  11.  
  12. {"common":{"ar":"110000","uid":"33","os":"Android 11.0","ch":"oppo","is_new":"0","md":"vivo iqoo3","mid":"mid_8","vc":"v2.1.132","ba":"vivo"},"page":{"page_id":"home","during_time":2347},"displays":[{"display_type":"activity","item":"1","item_type":"activity_id","pos_id":1,"order":1},{"display_type":"activity","item":"2","item_type":"activity_id","pos_id":1,"order":2},{"display_type":"query","item":"6","item_type":"sku_id","pos_id":2,"order":3},{"display_type":"query","item":"5","item_type":"sku_id","pos_id":3,"order":4},{"display_type":"promotion","item":"7","item_type":"sku_id","pos_id":1,"order":5},{"display_type":"recommend","item":"2","item_type":"sku_id","pos_id":5,"order":6},{"display_type":"query","item":"8","item_type":"sku_id","pos_id":2,"order":7},{"display_type":"query","item":"1","item_type":"sku_id","pos_id":3,"order":8},{"display_type":"query","item":"4","item_type":"sku_id","pos_id":4,"order":9},{"display_type":"query","item":"9","item_type":"sku_id","pos_id":4,"order":10},{"display_type":"query","item":"5","item_type":"sku_id","pos_id":4,"order":11},{"display_type":"query","item":"10","item_type":"sku_id","pos_id":1,"order":12}],"ts":1690880914000}
  13.  
  14. >>>>>>>>>>>> VisitorStats(stt=2023-08-01 16:57:10, edt=2023-08-01 16:57:20, vc=v2.1.132, ch=vivo, ar=310000, is_new=0, uv_ct=1, pv_ct=0, sv_ct=0, uj_ct=0, dur_sum=0, ts=1690880232000)
  15. >>>>>>>>>>>> VisitorStats(stt=2023-08-01 16:57:10, edt=2023-08-01 16:57:20, vc=v2.1.134, ch=oppo, ar=530000, is_new=0, uv_ct=1, pv_ct=0, sv_ct=0, uj_ct=0, dur_sum=0, ts=1690880233000)
  16. >>>>>>>>>>>> VisitorStats(stt=2023-08-01 17:07:20, edt=2023-08-01 17:07:30, vc=v2.1.134, ch=web, ar=310000, is_new=0, uv_ct=1, pv_ct=0, sv_ct=0, uj_ct=0, dur_sum=0, ts=1690880847000)
  17.

 120.

  1. Start the procedure:
  2.    BaseLogApp.java 
  3.    VisitorStatsApp.java 
  4.    UniqueVisitApp.java
  5.    UserJumpDetailApp.java 
  6.  
  7. Startup log:
  8.    java -jar gmall2020-mock-log-20-12-18.jar 
  9.    java -jar gmall-logger.jar

 

  1. 实例：
  2. UserJumpDetailApp:
  3. {"common":{"ar":"530000","uid":"50","os":"Android 11.0","ch":"oppo","is_new":"0","md":"Xiaomi 10 Pro ","mid":"mid_3","vc":"v2.1.134","ba":"Xiaomi"},"page":{"page_id":"home","during_time":19370},"displays":[{"display_type":"activity","item":"1","item_type":"activity_id","pos_id":5,"order":1},{"display_type":"activity","item":"2","item_type":"activity_id","pos_id":5,"order":2},{"display_type":"query","item":"8","item_type":"sku_id","pos_id":5,"order":3},{"display_type":"query","item":"2","item_type":"sku_id","pos_id":3,"order":4},{"display_type":"promotion","item":"8","item_type":"sku_id","pos_id":4,"order":5},{"display_type":"promotion","item":"6","item_type":"sku_id","pos_id":3,"order":6},{"display_type":"promotion","item":"5","item_type":"sku_id","pos_id":1,"order":7}],"ts":1690879690000}
  4. {"common":{"ar":"110000","uid":"24","os":"Android 11.0","ch":"oppo","is_new":"0","md":"Sumsung Galaxy S20","mid":"mid_19","vc":"v2.1.134","ba":"Sumsung"},"page":{"page_id":"home","during_time":12777},"displays":[{"display_type":"activity","item":"2","item_type":"activity_id","pos_id":4,"order":1},{"display_type":"activity","item":"2","item_type":"activity_id","pos_id":4,"order":2},{"display_type":"query","item":"4","item_type":"sku_id","pos_id":3,"order":3},{"display_type":"recommend","item":"8","item_type":"sku_id","pos_id":2,"order":4},{"display_type":"query","item":"5","item_type":"sku_id","pos_id":2,"order":5},{"display_type":"promotion","item":"6","item_type":"sku_id","pos_id":4,"order":6}],"ts":1690879770000}
  5. {"common":{"ar":"440000","uid":"20","os":"Android 11.0","ch":"vivo","is_new":"0","md":"vivo iqoo3","mid":"mid_7","vc":"v2.1.132","ba":"vivo"},"page":{"page_id":"home","during_time":13454},"displays":[{"display_type":"activity","item":"2","item_type":"activity_id","pos_id":1,"order":1},{"display_type":"activity","item":"2","item_type":"activity_id","pos_id":1,"order":2},{"display_type":"query","item":"4","item_type":"sku_id","pos_id":3,"order":3},{"display_type":"query","item":"2","item_type":"sku_id","pos_id":1,"order":4},{"display_type":"query","item":"1","item_type":"sku_id","pos_id":3,"order":5},{"display_type":"promotion","item":"3","item_type":"sku_id","pos_id":1,"order":6},{"display_type":"promotion","item":"9","item_type":"sku_id","pos_id":2,"order":7},{"display_type":"recommend","item":"5","item_type":"sku_id","pos_id":5,"order":8},{"display_type":"query","item":"5","item_type":"sku_id","pos_id":5,"order":9}],"ts":1690879766000}
  6.  
  7.  
  8. VisitorStatsApp:
  9. >>>>>>>>>>>> VisitorStats(stt=2023-08-01 16:49:30, edt=2023-08-01 16:49:40, vc=v2.1.134, ch=Appstore, ar=370000, is_new=0, uv_ct=0, pv_ct=2, sv_ct=1, uj_ct=0, dur_sum=6736, ts=1690879775000)
  10. >>>>>>>>>>>> VisitorStats(stt=2023-08-01 16:49:30, edt=2023-08-01 16:49:40, vc=v2.1.134, ch=vivo, ar=110000, is_new=0, uv_ct=0, pv_ct=2, sv_ct=1, uj_ct=0, dur_sum=30120, ts=1690879774000)
  11. >>>>>>>>>>>> VisitorStats(stt=2023-08-01 16:49:30, edt=2023-08-01 16:49:40, vc=v2.1.134, ch=oppo, ar=110000, is_new=0, uv_ct=0, pv_ct=1, sv_ct=1, uj_ct=1, dur_sum=12777, ts=1690879770000)

152-Flink实时数仓-DWS层-商品主题-整体测试

  1. 启动任务:
  2.    BaseDBApp.java
  3.    ods/FlinkCDC.java 
  4.    BaseLogApp.java
  5.    OrderWideApp.java
  6.    PaymentWideApp.java 
  7.    ProductStatsApp.java 
  8.  
  9. 创建表：
  10. create table product_stats_210325 (
  11.  stt DateTime,
  12.  edt DateTime,
  13.  sku_id UInt64,
  14.  sku_name String,
  15.  sku_price Decimal64(2),
  16.  spu_id UInt64,
  17.  spu_name String ,
  18.  tm_id UInt64,
  19.  tm_name String,
  20.  category3_id UInt64,
  21.  category3_name String ,
  22.  display_ct UInt64,
  23.  click_ct UInt64,
  24.  favor_ct UInt64,
  25.  cart_ct UInt64,
  26.  order_sku_num UInt64,
  27.  order_amount Decimal64(2),
  28.  order_ct UInt64 ,
  29.  payment_amount Decimal64(2),
  30.  paid_order_ct UInt64,
  31.  refund_order_ct UInt64,
  32.  refund_amount Decimal64(2),
  33.  comment_ct UInt64,
  34.  good_comment_ct UInt64 ,
  35.  ts UInt64
  36. )engine =ReplacingMergeTree(ts)
  37.  partition by toYYYYMMDD(stt)
  38.  order by (stt,edt,sku_id );
  39.  
  40.  
  41. 造数据：
  42.      注意：日期都改为当天 
  43.      java -jar gmall2020-mock-log-2020-12-18.jar
  44.      java -jar gmall-logger.jar
  45.  
  46.      java -jar gmall2020-mock-db-2020-11-27.jar 

  1. FlinkCDC.java 
  2. {"database":"gmall-210325-flink","before":{},"after":{"create_time":"2023-08-30 17:49:43","user_id":1017,"appraise":"1201","comment_txt":"评论内容：42324344452486777298996128427291877464398933868315","sku_id":15,"id":1696822496305336323,"spu_id":4,"order_id":27720},"type":"insert","tableName":"comment_info"}
  3. {"database":"gmall-210325-flink","before":{},"after":{"create_time":"2023-08-30 17:49:43","user_id":686,"appraise":"1201","comment_txt":"评论内容：23348134682275467263161282892337962443344368354913","sku_id":20,"id":1696822496309530626,"spu_id":6,"order_id":27724},"type":"insert","tableName":"comment_info"}
  4. {"database":"gmall-210325-flink","before":{},"after":{"create_time":"2023-08-30 17:49:43","user_id":3004,"appraise":"1204","comment_txt":"评论内容：49324881795631648686613552784978122797446563289775","sku_id":29,"id":1696822496309530627,"spu_id":10,"order_id":27724},"type":"insert","tableName":"comment_info"}
  5.  
  6. BaseDBApp.java 
  7. Kafka>>>>>>>>> {"sinkTable":"dwd_order_info_update","database":"gmall-210325-flink","before":{"delivery_address":"第11大街第21号楼8单元158门","consignee":"沈蕊","create_time":"2023-08-30 17:49:42","order_comment":"描述654765","expire_time":"2023-08-30 18:04:42","original_total_amount":20992.00,"coupon_reduce_amount":0.00,"order_status":"1002","out_trade_no":"131946498159351","total_amount":21005.00,"user_id":3151,"img_url":"http:img.gmall.com/196714.jpg","province_id":23,"feight_fee":13.00,"consignee_tel":"13100154190","trade_body":"Apple iPhone 12 (A2404) 128GB 黑色 支持移动联通电信5G 双卡双待手机等4件商品","id":27729,"activity_reduce_amount":0.00,"operate_time":"2023-08-30 17:49:42"},"after":{"delivery_address":"第11大街第21号楼8单元158门","consignee":"沈蕊","create_time":"2023-08-30 17:49:42","order_comment":"描述654765","expire_time":"2023-08-30 18:04:42","original_total_amount":20992.00,"coupon_reduce_amount":0.00,"order_status":"1004","out_trade_no":"131946498159351","total_amount":21005.00,"user_id":3151,"img_url":"http:img.gmall.com/196714.jpg","province_id":23,"feight_fee":13.00,"consignee_tel":"13100154190","trade_body":"Apple iPhone 12 (A2404) 128GB 黑色 支持移动联通电信5G 双卡双待手机等4件商品","id":27729,"activity_reduce_amount":0.00,"operate_time":"2023-08-30 17:49:43"},"type":"update","tableName":"order_info"}
  8.  
  9. BaseLogApp.java
  10. Display>>>>>>>>>>>>> {"display_type":"recommend","page_id":"home","item":"8","item_type":"sku_id","pos_id":2,"order":11}
  11. Page>>>>>>>>>>>> {"common":{"ar":"110000","uid":"35","os":"Android 11.0","ch":"xiaomi","is_new":"0","md":"Xiaomi Mix2 ","mid":"mid_15","vc":"v2.1.134","ba":"Xiaomi"},"page":{"page_id":"good_detail","item":"1","during_time":16024,"item_type":"sku_id","last_page_id":"home","source_type":"query"},"displays":[{"display_type":"promotion","item":"6","item_type":"sku_id","pos_id":4,  "order":1},{"display_type":"promotion","item":"3","item_type":"sku_id","pos_id":1,"order":2},{"display_type":"query","item":"5","item_type":"sku_id","pos_id":4,"order":3},{"display_type":"promotion","item":"8","item_type":"sku_id","pos_id":4,"order":4},{"display_type":"query","item":"9","item_type":"sku_id","pos_id":3,  "order":5},{"display_type":"query","item":"2","item_type":"sku_id","pos_id":5,"order":6}],"actions":[{"item":"1","action_id":"get_coupon","item_type":"coupon_id","ts":1693388957012}],"ts":1693388949000}
  12. Display>>>>>>>>>>>>> {"display_type":"promotion","page_id":"good_detail","item":"6","item_type":"sku_id","pos_id":4,"order":1}
  13. Display>>>>>>>>>>>>> {"display_type":"promotion","page_id":"good_detail","item":"3","item_type":"sku_id","pos_id":1,"order":2}
  14. Display>>>>>>>>>>>>> {"display_type":"query","page_id":"good_detail","item":"5","item_type":"sku_id","pos_id":4,"order":3}
  15. Display>>>>>>>>>>>>> {"display_type":"promotion","page_id":"good_detail","item":"8","item_type":"sku_id","pos_id":4,"order":4}
  16. Display>>>>>>>>>>>>> {"display_type":"query","page_id":"good_detail","item":"9","item_type":"sku_id","pos_id":3,"order":5}
  17. Display>>>>>>>>>>>>> {"display_type":"query","page_id":"good_detail","item":"2","item_type":"sku_id","pos_id":5,"order":6}
  18. Start>>>>>>>>>>>> {"common":{"ar":"500000","uid":"39","os":"Android 11.0","ch":"xiaomi","is_new":"0","md":"Huawei P30","mid":"mid_4","vc":"v2.1.134","ba":"Huawei"},"start":{"entry":"icon","open_ad_skip_ms":0,"open_ad_ms":3325,"loading_time":3394,"open_ad_id":3},"ts":1693388949000}
  19.  
  20. OrderWideApp.java 
  21. orderWideWithCategory3DS>>>>>>>>>>>> OrderWide(detail_id=82477, order_id=27728, sku_id=15, order_price=4488.00, sku_num=1, sku_name=华为 HUAWEI P40 麒麟990 5G SoC芯片 5000万超感知徕卡三摄 30倍数字变焦 8GB+128GB冰霜银全网通5G手机, province_id=13, order_status=1001, user_id=3018, total_amount=12712.00, activity_reduce_amount=0.00, coupon_reduce_amount=0.00, original_total_amount=12696.00, feight_fee=16.00,   split_feight_fee=null, split_activity_amount=null, split_coupon_amount=null, split_total_amount=4488.00, expire_time=null, create_time=2023-08-30 17:49:42, operate_time=null, create_date=2023-08-30, create_hour=17, province_name=重庆, province_area_code=500000, province_iso_code=CN-50, province_3166_2_code=CN-CQ, user_age=37, user_gender=F, spu_id=4,   tm_id=3, category3_id=61, spu_name=HUAWEI P40, tm_name=华为, category3_name=手机)
  22. orderWideWithCategory3DS>>>>>>>>>>>> OrderWide(detail_id=82479, order_id=27729, sku_id=12, order_price=9197.00, sku_num=2, sku_name=Apple iPhone 12 (A2404) 128GB 黑色 支持移动联通电信5G 双卡双待手机, province_id=23, order_status=1001, user_id=3151, total_amount=21005.00, activity_reduce_amount=0.00, coupon_reduce_amount=0.00, original_total_amount=20992.00, feight_fee=13.00, split_feight_fee=null,   split_activity_amount=null, split_coupon_amount=null, split_total_amount=18394.00, expire_time=null, create_time=2023-08-30 17:49:42, operate_time=null, create_date=2023-08-30, create_hour=17, province_name=河南, province_area_code=410000, province_iso_code=CN-41, province_3166_2_code=CN-HA, user_age=22, user_gender=M, spu_id=3, tm_id=2,   category3_id=61, spu_name=Apple iPhone 12, tm_name=苹果, category3_name=手机)
  23.  
  24. PaymentWideApp.java 
  25. >>>>>>>>>> PaymentWide(payment_id=18606, subject=金沙河面条 原味银丝挂面 龙须面 方便速食拉面 清汤面 900g等3件商品, payment_type=1102, payment_create_time=2023-08-30 17:49:42, callback_time=2023-08-30 17:50:02, detail_id=82477, order_id=27728, sku_id=15, order_price=4488.00, sku_num=1, sku_name=华为 HUAWEI P40 麒麟990 5G SoC芯片 5000万超感知徕卡三摄 30倍数字变焦 8GB+128GB冰霜银全网通5G手机, province_id=13, order_status=1001,   user_id=3018, total_amount=12712.00, activity_reduce_amount=0.00, coupon_reduce_amount=0.00, original_total_amount=12696.00, feight_fee=16.00, split_feight_fee=null, split_activity_amount=null, split_coupon_amount=null, split_total_amount=4488.00, order_create_time=2023-08-30 17:49:42, province_name=重庆, province_area_code=500000, province_iso_code=CN-50, province_3166_2_code=CN-CQ,   user_age=37, user_gender=F, spu_id=4, tm_id=3, category3_id=61, spu_name=HUAWEI P40, tm_name=华为, category3_name=手机)
  26. >>>>>>>>>> PaymentWide(payment_id=18607, subject=Apple iPhone 12 (A2404) 128GB 黑色 支持移动联通电信5G 双卡双待手机等4件商品, payment_type=1102, payment_create_time=2023-08-30 17:49:42, callback_time=2023-08-30 17:50:02, detail_id=82479, order_id=27729, sku_id=12, order_price=9197.00, sku_num=2, sku_name=Apple iPhone 12 (A2404) 128GB 黑色 支持移动联通电信5G 双卡双待手机, province_id=23, order_status=1001,   user_id=3151, total_amount=21005.00, activity_reduce_amount=0.00, coupon_reduce_amount=0.00, original_total_amount=20992.00, feight_fee=13.00, split_feight_fee=null, split_activity_amount=null, split_coupon_amount=null, split_total_amount=18394.00, order_create_time=2023-08-30 17:49:42, province_name=河南, province_area_code=410000, province_iso_code=CN-41, province_3166_2_code=CN-HA,   user_age=22, user_gender=M, spu_id=3, tm_id=2, category3_id=61, spu_name=Apple iPhone 12, tm_name=苹果, category3_name=手机)
  27.  
  28. ProductStatsApp.java 
  29. ProductStats(stt=2023-08-30 17:49:00, edt=2023-08-30 17:49:10, sku_id=2, sku_name=小米10 至尊纪念版 双模5G 骁龙865 120HZ高刷新率 120倍长焦镜头 120W快充 12GB+256GB 陶瓷黑 游戏手机, sku_price=6999, spu_id=1, spu_name=小米10, tm_id=5, tm_name=小米, category3_id=61, category3_name=手机, display_ct=27, click_ct=4, favor_ct=0, cart_ct=0,   order_sku_num=0, order_amount=0, order_ct=0, payment_amount=0, paid_order_ct=0, refund_order_ct=0, refund_amount=0, comment_ct=0, good_comment_ct=0, orderIdSet=[], paidOrderIdSet=[], refundOrderIdSet=[], ts=1693388941000)
  30. ProductStats(stt=2023-08-30 17:49:00, edt=2023-08-30 17:49:10, sku_id=8, sku_name=Apple iPhone 12 (A2404) 64GB 黑色 支持移动联通电信5G 双卡双待手机, sku_price=8197, spu_id=3, spu_name=Apple iPhone 12, tm_id=2, tm_name=苹果, category3_id=61, category3_name=手机, display_ct=36, click_ct=0, favor_ct=0, cart_ct=0, order_sku_num=0,   order_amount=0, order_ct=0, payment_amount=0, paid_order_ct=0, refund_order_ct=0, refund_amount=0, comment_ct=0, good_comment_ct=0, orderIdSet=[], paidOrderIdSet=[], refundOrderIdSet=[], ts=1693388940000)

 

157-DWS层-商品主题  代码编写  将数据写入ClickHouse&测试

  1. 启动kafka:
  2.    bin/kafka-server-start.sh config/server9092.properties 
  3.   
  4. 启动clickhouse:
  5.    docker exec -it clickhouse-server /bin/bash
  6.    clickhouse-client
  7.  
  8. 启动reids:
  9.    docker exec -it redis redis-cli
  10.  
  11. 启动程序:
  12.    BaseDBApp.java 
  13.    ods/FlinkCDC.java
  14.    OrderWideApp.java
  15.    ProvinceStatsSqlApp.java  
  16.  
  17. create table province_stats_210325 (
  18.  stt DateTime,
  19.  edt DateTime,
  20.  province_id UInt64,
  21.  province_name String,
  22.  area_code String,
  23.  iso_code String,
  24.  iso_3166_2 String,
  25.  order_amount Decimal64(2),
  26.  order_count UInt64,
  27.  ts UInt64
  28. )engine =ReplacingMergeTree(ts)
  29.  partition by toYYYYMMDD(stt)
  30.  order by (stt,edt,province_id);

标题

164-DWS层-关键词主题 代码测试

  1. 启动kafka:
  2.    bin/kafka-server-start.sh config/server9092.properties 
  3.   
  4. 启动clickhouse:
  5.    docker exec -it clickhouse-server /bin/bash
  6.    clickhouse-client
  7.  
  8. 启动程序:
  9.    KeywordStatsApp.java
  10.    BaseLogApp.java
  11.  
  12. create table keyword_stats_210325 (
  13.  stt DateTime,
  14.  edt DateTime,
  15.  keyword String,
  16.  source String,
  17.  ct UInt64,
  18.  ts UInt64
  19. )engine =ReplacingMergeTree(ts)
  20.  partition by toYYYYMMDD(stt)
  21.  order by (stt,edt,keyword,source);
  22.  
  23. 启动行为日志：
  24.     /opt/modules/gmall-flink/rt_applog
  25.     sudo java -jar gmall-logger.jar
  26.     sudo java -jar gmall2020-mock-log-2020-12-18.jar

gmall2020-mock-log-2020-12-18.jar 日志是不间断的造数据

每隔10秒(窗口大小是10秒)看数据的输出：

注意：clickhouse数据大小每次设置为5的倍数

 

  1. Start>>>>>>>>>>>> {"common":{"ar":"230000","uid":"18","os":"iOS 13.3.1","ch":"Appstore","is_new":"0","md":"iPhone 8","mid":"mid_15","vc":"v2.1.134","ba":"iPhone"},"start":{"entry":"icon","open_ad_skip_ms":0,"open_ad_ms":7254,"loading_time":13906,"open_ad_id":16},"ts":1608304326000}
  2.  
  3. Page>>>>>>>>>>>> {"common":{"ar":"230000","uid":"18","os":"iOS 13.3.1","ch":"Appstore","is_new":"0","md":"iPhone 8","mid":"mid_15","vc":"v2.1.134","ba":"iPhone"},"page":{"page_id":"home","during_time":5869},"displays":[{"display_type":"activity","item":"1","item_type":"activity_id","pos_id":1,"order":1},{"display_type":"activity","item":"1","item_type":"activity_id","pos_id":1,"order":2},{"display_type":"query","item":"5","item_type":"sku_id","pos_id":3,"order":3},{"display_type":"query","item":"7","item_type":"sku_id","pos_id":1,"order":4},{"display_type":"promotion","item":"4","item_type":"sku_id","pos_id":4,"order":5},{"display_type":"promotion","item":"1","item_type":"sku_id","pos_id":4,"order":6},{"display_type":"query","item":"4","item_type":"sku_id","pos_id":5,"order":7}],"ts":1608304326000}
  4.  
  5. Display>>>>>>>>>>>>> {"display_type":"activity","page_id":"home","item":"1","item_type":"activity_id","pos_id":1,"order":1}

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号