数据采集的flume架构

Posted 2021-06-24 海洋饼干1003

测试1:

新建一个flume1.conf文件

name

a1.sources = r1
a1.channels = c1
a1.sinks = k1

source

a1.sources.r1.type = netcat
a1.sources.r1.bind = DAQ102
a1.sources.r1.port = 6666

channel

a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 1000

sink

a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = atguigu
a1.sinks.k1.kafka.bootstrap.servers = DAQ102:9092
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.useFlumeEventFormat = false

bind

a1.sources.r1.channels=c1
a1.sinks.k1.channel=c1

测试2:
使用选择器,将不同的数据添加到不同的topic中

拦截器代码:

package com.hybg.daq;

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;

public class MyInterceptor implements Interceptor {

@Override
public void initialize() {

}

@Override
public Event intercept(Event event) {

    //原理:根据body的数据包含什么内容在header中添加什么内容
    //获取header文件
    Map<String, String> headers = event.getHeaders();
    //获取body文件
    String string = new String(event.getBody(), StandardCharsets.UTF_8);
    //判断是否包含某个文件
    if(string.contains("atguigu")){
        headers.put("topic","atguigu");
    }else if(string.contains("shangguigu")){
        headers.put("topic","shangguigu");
    }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 MyBuilder implements Builder{

    @Override
    public Interceptor build() {
        return new MyInterceptor();
    }

    @Override
    public void configure(Context context) {

    }
}

}

name

a1.sources = r1
a1.channels = c1
a1.sinks = k1

source

a1.sources.r1.type = netcat
a1.sources.r1.bind = DAQ102
a1.sources.r1.port = 4444

配置拦截器

a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.hybg.daq.MyInterceptor$MyBuilder

channel

a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 1000

sink

a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = other
a1.sinks.k1.kafka.bootstrap.servers = DAQ102:9092,DAQ103:9092,DAQ104:9092
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.useFlumeEventFormat = false

bind

a1.sources.r1.channels=c1
a1.sinks.k1.channel=c1

测试三:

name

a1.sources = r1
a1.channels = c1
a1.sinks = k1

source

a1.sources.source1.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.source1.batchSize = 5000
a1.sources.source1.batchDurationMillis = 2000
a1.sources.source1.kafka.bootstrap.servers = DAQ102:9092,DAQ103:9092,DAQ104:9092
a1.sources.source1.kafka.topics = atguigu,shangguigu,other
a1.sources.source1.kafka.consumer.group.id = customs

channel

a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 1000

sinks

a1.sinks.k1.type = logger

bind

a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

①三个组件都有

name

a1.sources = r1
a1.channels = c1
a1.sinks=k1

sources

a1.sources.r1.type = netcat
a1.sources.r1.bind = DAQ102
a1.sources.r1.port = 5555

channel

a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.c1.kafka.bootstrap.servers = DAQ102:9092,DAQ103:9092,DAQ104:9092

a1.channels.c1.kafka.topic = atguigu

a1.channels.c1.parseAsFlumeEvent = false

sink

a1.sinks.k1.type = logger

bind

a1.sources.r1.channels = c1
a1.sinks.r1.channel = c1

②有source有channel

name

a1.sources = r1
a1.channels = c1

sources

a1.sources.r1.type = netcat
a1.sources.r1.bind = DAQ102
a1.sources.r1.port = 5555

channel

a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.c1.kafka.bootstrap.servers = DAQ102:9092,DAQ103:9092,DAQ104:9092
a1.channels.c1.kafka.topic = atguigu
a1.channels.c1.parseAsFlumeEvent = false

bind

a1.sources.r1.channels = c1

③有sink有channel

name

a1.channels = c1
a1.sinks=k1

channel

a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.c1.kafka.bootstrap.servers = DAQ102:9092,DAQ103:9092,DAQ104:9092

a1.channels.c1.kafka.topic = atguigu

a1.channels.c1.parseAsFlumeEvent = false

sink

a1.sinks.k1.type = logger

bind

a1.sinks.r1.channel = c1

数据采集模块:

第一层flume:

name

a1.sources = r1
a1.channels = c1

sources

a1.sources.r1.type = TAILDIR
a1.sources.r1.filegroups = f1

将文件夹中所有的app的文件都进行读取

a1.sources.r1.filegroups.f1 = /opt/module/applog/log/app.*
a1.sources.r1.batchSize = 1000

设置断点续传的文件存储位置

a1.sources.r1.positionFile =/opt/module/flume-1.9.0/position/taildir_position.json

拦截器j

a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.hybg.daq.TaildirInterceptor$MyBuilder

channel

a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.c1.kafka.bootstrap.servers = DAQ102:9092,DAQ103:9092
a1.channels.c1.kafka.topic = topic_log
a1.channels.c1.parseAsFlumeEvent = false

bind

a1.sources.r1.channels = c1

第二层flume:

name

a2.sources = r1
a2.channels = c1
a2.sinks = k1

source

a2.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
a2.sources.r1.batchSize = 1000
a2.sources.r1.batchDurationMillis = 2000
a2.sources.r1.kafka.bootstrap.servers = DAQ102:9092,DAQ103:9092
a2.sources.r1.kafka.topics = topic_log
a2.sources.r1.useFlumeEventFormat = false

拦截器

设置拦截器后可能会导致错误,出现timestrap不能读取的问题

a2.sources.r1.interceptors = i1
a2.sources.r1.interceptors.i1.type = com.hybg.daq.timeInterceptor$MyBuilder

channel

a2.channels.c1.type = file

文件的

a2.channels.c1.dataDirs = /opt/module/flume-1.9.0/jobs/filechannel
a2.channels.c1.capacity = 1000000
a2.channels.c1.transactionCapacity = 10000
a2.channels.c1.checkpointDir = /opt/module/flume-1.9.0/jobs/checkpointdir
a2.channels.c1.keep-alive = 3

a1.channels.c1.useDualCheckpoints = true

a1.channels.c1. backupCheckpointDir = /otherdiskdir

sinks

a2.sinks.k1.type = hdfs
a2.sinks.k1.hdfs.path = /origin_data/gmall/log/topic_log/%Y-%m-%d
a2.sinks.k1.hdfs.filePrefix = log-
a2.sinks.k1.hdfs.round = false
a2.sinks.k1.hdfs.roundValue = 10
a2.sinks.k1.hdfs.rollSize = 134217728
a2.sinks.k1.hdfs.rollCount = 0

控制输出文件是原生文件。

a2.sinks.k1.hdfs.fileType = CompressedStream
a2.sinks.k1.hdfs.codeC = lzop

bind

a2.sources.r1.channels = c1
a2.sinks.k1.channel = c1

启动脚本2:

!/bin/bash

if [ $# -lt 1 ]
then
echo “<<<<<<<<<<<<<<<<<<<<输入有效参数>>>>>>>>>>>>>>>>>>>>”
echo “{start,stop}”
exit
fi

case $1 in
“start”)
echo “<<<<<<<<<<<<<<<<<<<<向HDFS传递数据>>>>>>>>>>>>>>>>>>>>”
ssh DAQ104 “flume-ng agent -c $FLUME_HOME/conf -f $FLUME_HOME/jobs/flume_2.conf -n a2 -Dflume.root.logger=INFO,console 1>/opt/module/flume-1.9.0 2>&1 &”
;;
“stop”)
echo “<<<<<<<<<<<<<<<<<<<<停止向HDFS传递数据>>>>>>>>>>>>>>>>>>>>”
ssh DAQ104 “ps -ef | grep flume_2.conf | grep -v grep | awk ‘{print $2}’ | xargs kill -9”
;;
*)
echo “<<<<<<<<<<<<<<<<<<<<参数错误>>>>>>>>>>>>>>>>>>>>”
;;
esac
————————————————
版权声明：本文为CSDN博主「海洋饼干1126」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/weixin_44868211/article/details/118104017