Data Warehouse: Real-Time, part Ⅰ

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;

//@Controller //注解标明 Controller
@RestController //@Controller + @ResponseBody
@Slf4j //lombok中的包
public class LoggerController{
    @RequestMapping("test2")
    public String test2(@RequestParam("name") String nn,
                       @RequestParam("age", defaultValue = "18") int age){
        return "success" + name + age;
    }
  
    @Autowired //自动注入
    private KafkaTemplate<String, String> kafkaTemplate;
  
    @RequestMapping("applog")
    public String test2(@RequestParam("param") String jsonStr){
        //sparkmall-mock：模拟生成数据模块
        //mock服务器不停的访问 http://pc_IP:8080/applog?jsonStr=* 产生原始数据
        /*1.修改SpringBoot核心配置：
        	server.port=8081
        	指定kafka代理地址，可以多个：spring.kafka.bootstrap-servers=hadoop102.9092
        	指定消息key和消息体的编解码方式：
        		spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerilizer 
        		spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerilizer*/
        /*2.Resources中添加 logback.xml 配置文件：决定如何打印、落盘、打印哪些日志 ，类似log4j
        	在注释@Slf4j后，使用log.info(jsonStr);log.warn(jsonStr);log.debug();log.error();log.trace()*/
        /*3.自动注入kafkaTemplate，producer send 数据，打开zk、kk服务
        	运行kafka消费者：bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic ods_base_log*/
        kafkaTemplate.send("ods_base_log", jsonStr);
        return "success";
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

public class FlinkCDC{
    public static void main(String[] args){
        //1.获取执行环境
        //1.1开启 CD 并指定后端为FS     memory fs rocksdb
        //1.2开启 checkpoint 实现断点续传
        //2.通过 flinkcsc 构建 sourcefunction 并读取数据
        //3.打印数据
        //4.启动任务
    }
}

1
2
3
4
5
6
7
8
9

public class FlihnkCDCWithSQL{
    public staic void main(String[] args){
        //1.获取执行环境
        //2.DDL方式建表
        //3.查询数据
        //4.将动态表转化为流
        //5.启动任务
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

@Override
public void deserialize(SourceRecord sourceRecord, Collector...){
    //1.创建Json对象用于存储最终数据
    //2.获取库名、表明
    //3.获取"before"数据
    //4.获取"after"数据
    //5.获取操作类型
    //6.写入JSON对象
    //7.输出数据
}

1
2
3

数据流：：web/app -> Nginx -> SpringBoot -> Mysql -> FlinkApp -> Kafka(ods) -> FlinkApp -> Kafka/Hbase(dwd-dim)

程序：mocklog -> Nginx -> Logger.sh -> KafkaZ(zk) -> BaseLogApp -> Kafka

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

public class BaseLogApp{
public static void main(String[] args){
    //1.获取执行环境
    //2.消费 ods_base_log 主题数据创建流
    //3.将每行数据转换为JSON对象
    	//3.1发生异常，如不满足JSON格式，捕获写入侧输出流
    //4.新老用户校验，状态编辑
    	//4.1 转态编程
    jsonObjDS.keyBy(...).map(new RichMapFunction<JSONObject, JSONObject>()){
        private ValueState<String> valueState;
        @Override
        public void open(Configuration parameters) throws Exception {
            valueState = getRuntimeContext().getState(new ValueStateDescriptor<String>("value-state", String.class));
        }
        @Override
        public JSONObject map(JSONObject value) throws Excepiton {
            //获取JSON中"is_new"标记
            String isNew = value.getJSONObject("common").getString("is_new");
            //isnew做处理，旧数据不处理
            if("1".equals(isNew)){
                String state = valueState.value();
                if(state != null){
                    //修改isnew标记
                    value.getJSONObject("common").put("is_new", "0");
                }else{
                    valueState.update("1");
                }
            }
            return value;
        }
    };
    //5.分流、侧输出流  页面：主流  启动：侧输出流  曝光：侧输出流
    //6.提取侧输出流
    //7.将三个流进行打印并输出到对应的kafka主题中
}
}

1
2
3

数据流：：web/app -> Nginx -> SpringBoot -> Mysql -> FlinkApp -> Kafka(ods) -> FlinkApp -> Kafka/Hbase(dwd-dim)

程序：            mockDB -> Mysql -> FlinkCDC -> Kafka(ZK) -> BaseDBApp -> Kafka/Hbase(Phoenix, zk, hdfs)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

BaseDBApp;
        //1.获取执行环境
        //2.消费 kafka ods_base_db 主题数据创建流
        //3.将每行数据转换成JSON对象并过滤 主流
        //4.使用FlinkCDC消费配置表并处理成 广播流
        //5.连接主流和广播流
        //6.分流 处理数据 广播流数据和主流数据(根据广播流数据进行处理)
        //7.提取kafka流数据和HBASE流数据
        //8.将kafka数据写入Kafka主题，将HBASE数据写入Phoenix表
        //9.启动