需求

需求就是将流量数据（json格式）中某个接口数据抽取一下。如：有个identityUri="yiyang/user/getById/13782" , 这里的13782，是个userId，我们需要将其处理成 identityUri="yiyang/user/getById/{}"

关于接口抽取，有两种方式：

1. 使用正则替换。正则替换不全，而且有风险，可能会被误替换
2. 如果能拿到swagger的接口列表，我们可以根据前缀树算法来进行匹配替换（推荐）。这种不会被误替换，但是如果增加接口，我们需要更新swagger的接口数据

实际上我们生产中是将二者接口使用的。先使用2，如果没有匹配到，在使用1

这里是演示flink kafka的用法，我们简单使用正则处理

依赖jar包

<properties>
        <flink.version>1.12.0</flink.version>
        <java.version>1.8</java.version>
        <scala.binary.version>2.11</scala.binary.version>
    </properties>

    <dependencies>
        <!--  kafka连接器 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.75</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
            <version>1.12.2</version>
        </dependency>
        <dependency>
            <artifactId>flink-clients_${scala.binary.version}</artifactId>
            <groupId>org.apache.flink</groupId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>1.2.3</version>
        </dependency>
    </dependencies>

编写代码

package com.liufei.flink;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class ConsumeKafkaTest {

    private static final Logger log = LoggerFactory.getLogger(ConsumeKafkaTest.class);

    public static void main(String[] args) {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // checkpoint
        // env.enableCheckpointing(10 * 1000L);

        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers", "192.168.18.144:9092"); // kafka地址
        prop.setProperty("group.id", "consumer_flink");

        // 消费消息的topic
        String consumeTopic = "yiyang";
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
                consumeTopic, new SimpleStringSchema(), prop);
        // 从最新的数据开始消费
        // kafkaConsumer.setStartFromLatest();
        kafkaConsumer.setStartFromGroupOffsets();

        // sink的topic
        String produceTopic = "yiyang_sink";
        FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>(
                produceTopic, new SimpleStringSchema(), prop);

        env.addSource(kafkaConsumer)
                .flatMap(new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String content, Collector<String> collector) throws Exception {
                        log.info("Flink msg: {}", content);
                        JSONObject jsonObject = JSONObject.parseObject(content);
                        jsonObject.put("identityUri", replaceUri(jsonObject.getString("identityUri")));
                        String contentSink = jsonObject.toJSONString();
                        log.info("Flink sink: {}", contentSink);
                        collector.collect(contentSink);
                    }
                })
                .addSink(kafkaProducer)
                .setParallelism(2);

        try {
            env.execute("My Flink Test");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 接口抽取
     * @param identityUri
     * @return
     */
    private static String replaceUri(String identityUri) {
        if (identityUri == null) {
            return null;
        }
        return identityUri.replaceAll("[0-9]+", "{}");
    }
}

注意：kafka消费的方式是： kafkaConsumer.setStartFromGroupOffsets();

• 从最早位点开始消费

consumer.setStartFromEarliest();

• 从指定时间点开始消费

consumer.setStartFromTimestamp(1559801580000l);

• 从最新的数据开始消费

# 如果消费组中之前有没有消费的消息，则不会被消费，会重置offset成最新的值，这种情况会导致消息丢失。一般不用这个，用下面的
consumer.setStartFromLatest();

2022-07-03 18:31:07.029 INFO  PID_IS_UNDEFINED --- [Kafka Fetcher for Source: Custom Source -> Flat Map (8/8)#0] o.a.k.clients.consumer.internals.SubscriptionState - [Consumer clientId=consumer-consumer_flink-15, groupId=consumer_flink] Resetting offset for partition yiyang-0 to offset 22.
2022-07-03 18:31:12.093 INFO  PID_IS_UNDEFINED --- [Kafka Fetcher for Source: Custom Source -> Flat Map (8/8)#0] o.a.k.c.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-consumer_flink-15, groupId=consumer_flink] Discovered group coordinator 192.168.18.144:9092 (id: 2147483647 rack: null)

看下上面的启动日志，有这样的信息：Resetting offset for partition yiyang-0 to offset 22.

• 从上次消费位点开始消费

consumer.setStartFromGroupOffsets();

验证

我们另外启动一个程序，发送消息，并消费两个topic中的数据

{"ip":"127.0.0.5","identityUri":"yiyang/user/getById/13782"}

看下 ConsumeKafkaTest 中的日志

2022-07-03 18:58:48.594 INFO  PID_IS_UNDEFINED --- [Legacy Source Thread - Source: Custom Source -> Flat Map (8/8)#0] com.liufei.flink.ConsumeKafkaTest - Flink msg: {"ip":"127.0.0.5","identityUri":"yiyang/user/getById/13782"}
2022-07-03 18:58:48.655 INFO  PID_IS_UNDEFINED --- [Legacy Source Thread - Source: Custom Source -> Flat Map (8/8)#0] com.liufei.flink.ConsumeKafkaTest - Flink sink: {"ip":"127.0.0.5","identityUri":"yiyang/user/getById/{}"}

在看下另外一个服务（消费两个topic数据）的日志：

image.png

说明已经成功的把处理好的消息发送到另外一个topic中了

扩展

关于数据处理，如果只是简单的增加字段，减少字段，正则替换，也可以使用logstash工具