Data Warehouse: Real-Time, part Ⅰ

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";
    }
}

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主题中
}
}