flink-sql对kafka数据进行清洗过滤

今天这篇blog主要记录使用flink-sql对kafka中的数据进行过滤。

以前对kafka数据进行实时处理时都是使用java来进行flink开发，需要创建一个工程，并且打成jar包再提交，流程固定但对于简单任务来说还是比较繁琐的。

今天我们要对logstash采集到kafka中的数据进行过滤筛选，将筛选后的数据发送给另外一个kafka topic，由于处理逻辑比较简单，使用flink自带的sql函数就可以搞定，所以我们今天就用flink-sql来解决这问题。

问题描述

我们需要筛选出ServiceA、ServiceB、ServiceC、ServiceD四个类打印出来的日志信息，并将目标信息发送到另外一个kafka topic。logstash推送到kafka中的日志格式如下，日志信息均在message字段中。

{"@version":"1","@timestamp":"2022-11-18T08:11:33.000Z","host":"localhost","message":"ServiceX XXXX","uid":3081609001,"type":"xxx"}

环境说明

flink 1.13.6

重要文档

flink-sql内置函数官方文档

flink kafka connector官方文档

实现代码

--sourceTableCREATETABLE omg_log(
    message VARCHAR)WITH('connector'='kafka','topic'='source-topic','properties.bootstrap.servers'='localhost:9092','properties.group.id'='group_id','properties.security.protocol'='SASL_PLAINTEXT','properties.sasl.mechanism'='PLAIN','properties.sasl.jaas.config'='org.apache.kafka.common.security.plain.PlainLoginModule required username="username" password="password";','scan.startup.mode'='group-offsets','format'='json','json.ignore-parse-errors'='true');--sinkTableCREATETABLE omg_log_sink (
    message VARCHAR)WITH('connector'='kafka','topic'='target-topic','properties.bootstrap.servers'='loaclhost:9093','properties.security.protocol'='SASL_PLAINTEXT','properties.sasl.mechanism'='PLAIN','properties.sasl.jaas.config'='org.apache.kafka.common.security.plain.PlainLoginModule required username="username" password="password";','format'='csv');--filter and insert INSERTINTO omg_log_sink(message)SELECT message
FROM omg_log
whereREGEXP(message,'ServiceA|ServiceB|ServiceC|ServiceD');