Kafka+Fink 实战+工具类

• LogServiceImpl
  

1. @Service
2. @Slf4j
3. public class LogServiceImpl implements LogService {
5.     private static final String TOPIC_NAME = "ods_link_visit_topic";
7.     @Autowired
8.     private KafkaTemplate kafkaTemplate;
10.     /**
11.      * 记录日志
12.      *
13.      * @param request
14.      * @param shortLinkCode
15.      * @param accountNo
16.      * @return
17.      */
18.     @Override
19.     public void recodeShortLinkLog(HttpServletRequest request, String shortLinkCode, Long accountNo) {
20.         // ip、 浏览器信息
21.         String ip = CommonUtil.getIpAddr(request);
22.         // 全部请求头
23.         Map<String, String> headerMap = CommonUtil.getAllRequestHeader(request);
25.         Map<String,String> availableMap = new HashMap<>();
26.         availableMap.put("user-agent",headerMap.get("user-agent"));
27.         availableMap.put("referer",headerMap.get("referer"));
28.         availableMap.put("accountNo",accountNo.toString());
30.         LogRecord logRecord = LogRecord.builder()
31.                 //日志类型
32.                 .event(LogTypeEnum.SHORT_LINK_TYPE.name())
33.                 //日志内容
34.                 .data(availableMap)
35.                 //客户端ip
36.                 .ip(ip)
37.                 // 时间
38.                 .ts(CommonUtil.getCurrentTimestamp())
39.                 //业务唯一标识（短链码）
40.                 .bizId(shortLinkCode).build();
42.         String jsonLog = JsonUtil.obj2Json(logRecord);
44.         //打印日志 in 控制台
45.         log.info(jsonLog);
47.         // 发送kafka
48.         kafkaTemplate.send(TOPIC_NAME,jsonLog);
51.     }
52. }

复制代码

• DwdShortLinkLogApp
  

1. @Slf4j
2. public class DwdShortLinkLogApp {
3.     //定义 topic
4.     public static final String SOURCE_TOPIC = "ods_link_visit_topic";
6.     //定义 消费组
7.     public static final String SINK_TOPIC = "dwd_link_visit_topic";
9.     //定义 消费组
10.     public static final String GROUP_ID = "dwd_short_link_group";
13.     public static void main(String[] args) throws Exception {
15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
17.         env.setParallelism(1);
19. //        DataStream<String> ds = env.socketTextStream("192.168.75.146", 8888);
21.         FlinkKafkaConsumer<String> kafkaConsumer = KafkaUtil.getKafkaConsumer(SOURCE_TOPIC, GROUP_ID);
23.         DataStreamSource<String> ds = env.addSource(kafkaConsumer);
25.         ds.print();
27.         SingleOutputStreamOperator<JSONObject> jsonDs = ds.flatMap(new FlatMapFunction<String, JSONObject>() {
29.             @Override
30.             public void flatMap(String value, Collector<JSONObject> out) throws Exception {
31.                 JSONObject jsonObject = JSON.parseObject(value);
32.                 // 生成web端设备唯一标识
33.                 String udid = getDeviceId(jsonObject);
34.                 jsonObject.put("udid",udid);
36.                 String referer = getReferer(jsonObject);
37.                 jsonObject.put("referer",referer);
39.                 out.collect(jsonObject);
41.             }
42.         });
44.         // 分组
45.         KeyedStream<JSONObject, String> keyedStream = jsonDs.keyBy(new KeySelector<JSONObject, String>() {
47.             @Override
48.             public String getKey(JSONObject value) throws Exception {
49.                 return value.getString("udid");
51.             }
52.         });
55.         // 识别新老访客    richMap open函数，对状态以及日期格式进行初始化
57.         SingleOutputStreamOperator<String> jsonDSWithVisitorState = keyedStream.map(new VisitorMapFunction());
59.         jsonDSWithVisitorState.print("ods新老访客");
61.         // 存储到dwd
62.         FlinkKafkaProducer<String> kafkaProducer = KafkaUtil.getKafkaProducer(SINK_TOPIC);
64.         jsonDSWithVisitorState.addSink(kafkaProducer);
67.         env.execute();
68.     }
70.     /**
71.      * 获取referer
72.      * @param jsonObject
73.      * @return
74.      */
75.     public static String getReferer(JSONObject jsonObject){
76.         JSONObject dataJsonObj = jsonObject.getJSONObject("data");
77.         if(dataJsonObj.containsKey("referer")){
79.             String referer = dataJsonObj.getString("referer");
80.             if(StringUtils.isNotBlank(referer)){
81.                 try {
82.                     URL url = new URL(referer);
83.                     return url.getHost();
84.                 } catch (MalformedURLException e) {
85.                     log.error("提取referer失败：{}",e.toString());
86.                 }
87.             }
88.         }
90.         return "";
92.     }
94.     /**
95.      * 生成设备唯一标识
96.      *
97.      * @param jsonObject
98.      * @return
99.      */
100.     public static String getDeviceId(JSONObject jsonObject){
101.         Map<String,String> map= new TreeMap<>();
103.         try{
104.             map.put("ip",jsonObject.getString("ip"));
105.             map.put("event",jsonObject.getString("event"));
106.             map.put("bizId",jsonObject.getString("bizId"));
107.             map.put("userAgent",jsonObject.getJSONObject("data").getString("userAgent"));
109.             return DeviceUtil.geneWebUniqueDeviceId(map);
111.         }catch (Exception e){
112.             log.error("生产唯一deviceId异常：{}", jsonObject);
113.             return null;
114.         }
117.     }
120. }

复制代码

• KafkaUtil
  1. @Slf4j
  2. public class KafkaUtil {
  4.     /**
  5.      * kafka 的 broker 地址
  6.      */
  7.     private static String KAFKA_SERVER = null;
  9.     static {
  10.         Properties properties = new Properties();
  12.         InputStream in = KafkaUtil.class.getClassLoader().getResourceAsStream("application.properties");
  14.         try {
  15.             properties.load(in);
  16.         } catch (IOException e) {
  17.             e.printStackTrace();
  18.             log.error("加载Kafka配置文件失败:{}",e.getMessage());
  19.         }
  21.         //获取配置文件中的value
  22.         KAFKA_SERVER = properties.getProperty("kafka.servers");
  24.     }
  26.     /**
  27.      * 获取flink的kafka消费者
  28.      * @param topic
  29.      * @param groupId
  30.      * @return
  31.      */
  32.     public static FlinkKafkaConsumer<String> getKafkaConsumer(String topic,String groupId){
  33.         Properties properties = new Properties();
  34.         properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG,groupId);
  35.         properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,KAFKA_SERVER);
  37.         return new FlinkKafkaConsumer<String>(topic,new SimpleStringSchema(),properties);
  38.     }
  40.     /**
  41.      * 获取flink的kafka生产者
  42.      * @param topic
  43.      * @return
  44.      */
  45.     public static FlinkKafkaProducer<String> getKafkaProducer(String topic){
  46.         return new FlinkKafkaProducer<String>(KAFKA_SERVER,topic,new SimpleStringSchema());
  47.     }
  48. }

  复制代码
• TimeUtil
  1. public class TimeUtil {
  3.     /**
  4.      * 默认日期格式
  5.      */
  6.     private static final String DEFAULT_PATTERN = "yyyy-MM-dd";
  8.     /**
  9.      * 默认日期格式
  10.      */
  11.     private static final DateTimeFormatter DEFAULT_DATE_TIME_FORMATTER  = DateTimeFormatter.ofPattern(DEFAULT_PATTERN);
  13.     private static final ZoneId DEFAULT_ZONE_ID = ZoneId.systemDefault();
  16.     /**
  17.      * LocalDateTime 转 字符串，指定日期格式
  18.      * @param localDateTime
  19.      * @param pattern
  20.      * @return
  21.      */
  22.     public static String format(LocalDateTime localDateTime, String pattern){
  23.         DateTimeFormatter formatter = DateTimeFormatter.ofPattern(pattern);
  24.         String timeStr = formatter.format(localDateTime.atZone(DEFAULT_ZONE_ID));
  25.         return timeStr;
  26.     }
  29.     /**
  30.      * Date 转 字符串, 指定日期格式
  31.      * @param time
  32.      * @param pattern
  33.      * @return
  34.      */
  35.     public static String format(Date time, String pattern){
  36.         DateTimeFormatter formatter = DateTimeFormatter.ofPattern(pattern);
  37.         String timeStr = formatter.format(time.toInstant().atZone(DEFAULT_ZONE_ID));
  38.         return timeStr;
  39.     }
  41.     /**
  42.      *  Date 转 字符串，默认日期格式
  43.      * @param time
  44.      * @return
  45.      */
  46.     public static String format(Date time){
  48.         String timeStr = DEFAULT_DATE_TIME_FORMATTER.format(time.toInstant().atZone(DEFAULT_ZONE_ID));
  49.         return timeStr;
  50.     }
  52.     /**
  53.      * timestamp 转 字符串，默认日期格式
  54.      *
  55.      * @param timestamp
  56.      * @return
  57.      */
  58.     public static String format(long timestamp) {
  59.         String timeStr = DEFAULT_DATE_TIME_FORMATTER.format(new Date(timestamp).toInstant().atZone(DEFAULT_ZONE_ID));
  60.         return timeStr;
  61.     }
  64.     /**
  65.      * 字符串 转 Date
  66.      *
  67.      * @param time
  68.      * @return
  69.      */
  70.     public static Date strToDate(String time) {
  71.         LocalDateTime localDateTime = LocalDateTime.parse(time, DEFAULT_DATE_TIME_FORMATTER);
  72.         return Date.from(localDateTime.atZone(DEFAULT_ZONE_ID).toInstant());
  74.     }
  76. }

  复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！