解决flume采集日志使用KafkaChannel写不到hdfs的问题

问题：

在hadoop102中执行lg.sh命令（一个脚本，可以生成日志）后，发现日志没有传输到hdfs中

解决：

第一步：

首先，先观察一下我们的zookeeper,kafka,flume,hadoop集群是否启动，要把这些集群都打开来。

第二步：

检查一下消费flume所在的集群，也就是hadoop104中的/opt/module/flume/conf目录下的kafka-flume-hdfs.conf文件

这是我的配置文件信息。供大家参考

## 组件
a1.sources=r1
a1.channels=c1
a1.sinks=k1
#配置kafkasource
## source1
a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.r1.batchSize = 5000
a1.sources.r1.batchDurationMillis = 2000
a1.sources.r1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092
a1.sources.r1.kafka.topics=topic_log
#配置时间拦截器
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.atguigu.flume.interceptor.TimeStampInterceptor$Builder
 
#配置filechannel
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /opt/module/flume/checkpoint/behavior1
a1.channels.c1.dataDirs = /opt/module/flume/data/behavior1/
 
 
##配置sink1
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /origin_data/gmall/log/topic_log/%Y-%m-%d
a1.sinks.k1.hdfs.filePrefix = log-
a1.sinks.k1.hdfs.round = false
 
#控制生成的小文件
a1.sinks.k1.hdfs.rollInterval = 10
a1.sinks.k1.hdfs.rollSize = 134217728
a1.sinks.k1.hdfs.rollCount = 0
 
## 控制输出文件是原生文件。
a1.sinks.k1.hdfs.fileType = CompressedStream
a1.sinks.k1.hdfs.codeC = lzop
 
## 拼装
a1.sources.r1.channels = c1
a1.sinks.k1.channel= c1

这里注意：在配置时间拦截器的一部分中，我的引用名是com.atguigu.flume.interceptor.TimeStampInterceptor

也就是我创建后面的maven工程时，用的包名（idea软件包）

这里，大家的包名可能不一样，根据自己创建的包名进行修改即可。

第三步：

创建maven工程，时间拦截器。

在com.atguigu.flume.interceptor包下创建TimeStampInterceptor类，然后实现如下代码

package com.atguigu.flume.interceptor;
 
import com.alibaba.fastjson.JSONObject;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
 
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
 
public class TimeStampInterceptor implements Interceptor {
 
    private ArrayList events = new ArrayList<>();
 
    @Override
    public void initialize() {
 
    }
 
    @Override
    public Event intercept(Event event) {
 
        Map headers = event.getHeaders();
        String log = new String(event.getBody(), StandardCharsets.UTF_8);
 
        JSONObject jsonObject = JSONObject.parseObject(log);
 
        String ts = jsonObject.getString("ts");
        headers.put("timestamp", ts);
 
        return event;
    }
 
    @Override
    public List intercept(List list) {
        events.clear();
        for (Event event : list) {
            events.add(intercept(event));
        }
 
        return events;
    }
 
    @Override
    public void close() {
 
    }
 
    public static class Builder implements Interceptor.Builder {
        @Override
        public Interceptor build() {
            return new TimeStampInterceptor();
        }
 
        @Override
        public void configure(Context context) {
        }
    }
}

然后将代码打包

打包过程：

点击idea右端的maven，双击package，即可打包

 打包后，点击以下步骤

 这时候，会弹出一个目录

（重点，我就是这里出错了） 然后将这个jar包放到hadoop104的/opt/module/flume/lib文件夹下面，然后分发给hadoop102,hadoop103中。

最后，重新启动一下集群，就可以了。

---

附：（抛出一下异常，说明你的消费flume内存不够，只要修改配置就可以了）