2022-08-10 Flume和Kafka对接_背景:flume将bdp端kafka数据传到udp端kafka存在延迟,其中,flume采用kafk

一、对接相关的Flume组件

1. Kafka Source

Kafka Source 会自动在Event Header中添加属性：timestamp、topic、partition、key。如果不想使用其自动添加timestamp，那么就需要自己在Header中覆盖。

2. Kafka Channel

Kafka Channel的功能很强大，用法有如下三种：

        (1) 当一个普通的缓冲Channel使用；

        (2) 只需要有一个Source和一个Kafka Channel就可以直接向kafka写入消息了；

        (3) 只需要有一个Kafka Channel和一个Sink就可以直接从kafka中读取数据并写出了；

3. Kafka Sink

Kafka Sink uses the topic and key properties from the FlumeEvent headers to send events to Kafka. If topic exists in the headers, the event will be sent to that specific topic, overriding the topic configured for the Sink. If key exists in the headers, the key will used by Kafka to partition the data between the topic partitions. Events with same key will be sent to the same partition. If the key is null, events will be sent to random partitions.

二、案例

1. 需求：

已知某一服务会一直向端口发送网址数据，现在要求收集这些网址，并将其发往至kafka中的websit和other主题中，website和other主题都具有两个分区，将含有baidu的网址存储在0号分区，将含有iqiyi的网址存储在1号分区，其他网址信息都存储在other主题中；

2. 架构图

3. 实现

(1) 引入依赖

    <dependencies>
        <dependency>
            <groupId>org.apache.flume</groupId>
            <artifactId>flume-ng-core</artifactId>
            <version>1.9.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.4.1</version>
        </dependency>
    </dependencies>

(2) 编写WebsiteInterceptor

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
 
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
 
public class WebsiteInterceptor implements Interceptor {
    @Override
    public void initialize() {
 
    }
 
    @Override
    public Event intercept(Event event) {
        Map<String, String> headers = event.getHeaders();
        String website = new String(event.getBody(), StandardCharsets.UTF_8);
        Pattern pattern = Pattern.compile("(baidu|iqiyi)");
        Matcher matcher = pattern.matcher(website);
        if (matcher.find()){
            headers.put("topic", "website");
            if (matcher.group().equalsIgnoreCase("baidu")){
                headers.put("key", "0");
            }else {
                headers.put("key", "1");
            }
        }else {
            headers.put("topic", "other");
        }
        return event;
    }
 
    @Override
    public List<Event> intercept(List<Event> list) {
        for (Event event : list) {
            intercept(event);
        }
        return list;
    }
 
    @Override
    public void close() {
 
    }
 
    public static class Builder implements Interceptor.Builder{
 
        @Override
        public Interceptor build() {
            return new WebsiteInterceptor();
        }
 
        @Override
        public void configure(Context context) {
 
        }
    }
}

(3) 编写配置文件

# 配置文件命名为：netcat-kafkasink.conf,放置于$FLUME_HOME/conf-files/netcat-kafkasink目录下
 
# 配置文件内容为：
# Named
a1.sources = r1
a1.channels = c1
a1.sinks = k1
 
# Source
# 对r1的配置
a1.sources.r1.type = netcat
a1.sources.r1.bind = hadoop101
a1.sources.r1.port = 6666
 
 
# Interceptors
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.fig.flume.interceptor.WebsiteInterceptor$Builder
 
# Channel Selector
...
 
# Channel
# 对c1的配置
a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 1000
 
 
# Sink Processor
...
 
# Sink
# 对k1的配置
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.bootstrap.servers = hadoop101:9092,hadoop102:9092,hadoop103:9092
a1.sinks.k1.flumeBatchSize = 1000
a1.sinks.k1.kafka.producer.acks = -1
a1.sinks.k1.useFlumeEventFormat = false
 
 
# Bind
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

(4) 启动 

# 开启三个消费者消费数据
kafka-ops.sh consumer --topic website --group tp --from-beginning
kafka-ops.sh consumer --topic website --group tp --from-beginning
kafka-ops.sh consumer --topic other --group tp --from-beginning
 
# 先启动agent
flume-ng agent -c $FLUME_HOME/conf -f $FLUME_HOME/conf-files/netcat-kafkasink/netcat-kafkasink.conf -n a1 -Dflume.root.logger=INFO,console
 
# 模拟向端口发送数据
nc hadoop101 6666