前端埋点数据采集（三）Flume采集数据

mv flume-env.sh.template flume-env.sh
# Enviroment variables can be set here.
#JAVA_HOME路径
export JAVA_HOME=/opt/module/jdk1.8.0_14

vim taildir-kafkachannel-kafka.conf

#flume配置信息
#Source:Taildir Source
#channel:kafka channel
#sink没有，直接写入到kafka
a1.sources = r1
a1.channels = c1
a1.sources.r1.type = TAILDIR
a1.sources.r1.channels = c1
a1.sources.r1.positionFile = opt/module/flume-1.9.0/taildir_position.json
a1.sources.r1.filegroups = f1


#/opt/module/applog/log/app.2021-04-30.log
a1.sources.r1.filegroups.f1 = opt/module/applog/log/app.*
#配置自定义的拦截器
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type= com.wqs.flume.interceptor.ETLInterceptor$Builder

#描述channel
a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
#flume采集是node01、02没有03，03后面做消费kafka.
a1.channels.c1.kafka.bootstrap.servers = node01:9092,node02:9092
#创建kafka的topic名字叫topic_log
a1.channels.c1.kafka.topic = topic_log
a1.channels.c1.parseAsFlumeEvent = false

<dependency>
    <groupId>org.apache.flume</groupId>
    <artifactId>flume-ng-core</artifactId>
    <version>1.9.0</version>
    <!--定义范围为只在编译使用-->
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.2.62</version>
</dependency>

/**
 * 验证数据是否是JSON
 * @param log
 * @return
 */
public static boolean isJSONValidate(String log)
{
    //加上异常处理，正常解析返回true,异常抛出
    try
    {
        JSON.parse(log);
        return true;
    }catch (JSONException e)
    {
        return false;
    }
}

public Event intercept(Event event)
{
  //获取Body内容
  byte[] body = event.getBody();
  //body转化为String,且编码是UTF-8
  String log = new String(body, StandardCharsets.UTF_8);
  //调用JSONUtils类的isValidate方法对其log校验
  if(JSONUtils.isJSONValidate(log))
  {
      //标准JSON，返回event
      return event;
  }
  else
  {
      //不是标准JSON，就清理脏数据，在哪里可以清理？集合里面清理
      return null;
  }
}

public List<Event> intercept(List<Event> list)
{
    //使用迭代器来清理非标准JSON
    Iterator iterator = list.iterator();
    while (iterator.hasNext())
    {
        //取出下一个event
        Event next = (Event) iterator.next();
        //调用单个event方法，返回null说明是非标准JSON
        if(intercept(next) == null)
        {
            //如果为空，则表示它是非标准JSON去除它remove
            iterator.remove();
        }
    }
    return list;
}

bin/flume-ng agent --name a1 --conf-file conf/taildir-kafkachannel-kafka.conf

bin/flume-ng agent --name a1 --conf-file conf/taildir-kafkachannel-kafka.conf &

no hup bin/flume-ng agent --name a1 --conf-file conf/taildir-kafkachannel-kafka.conf &

 #function:flume Cluster 采集log的 start and stop
#!bin/bash
case $1 in
"start"){
        for node in node01 node02
        do
                echo "===============启动 $node 采集flume ==============="      
                ssh $node "nohup opt/module/flume-1.9.0/bin/flume-ng agent --name a1 --conf-file opt/module/flume-1.9.0/conf/taildir-kafkachannel-kafka.conf --name a1 Dflume.root.logger=INFO,LOGFILE 1>/opt/module/flume-1.9.0/log1.txt 2>&1 &"
        done


};;
"stop"){
        for node in node01 node02
        do
                echo "===============停止 $node 采集flume ===============" 
                ssh $node "ps -ef | grep taildir-kafkachannel-kafka | grep -v grep |awk  '{print \$2}' | xargs -n1 kill -9 "
        done
};;
esac                                                                                                                                   

>>>>

Q&A

Caused by: 
cdjava.lang.ClassNotFoundException: com.google.common.collect.Lists
        at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
        at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
        ... 1 more

vim /etc/profile.d/my_env.sh
#HADOOP_HOME
export HADOOP_HOME=/opt/module/hadoop-3.1.4
export PATH=$PATH:$HADOOP_HOME/bin
export PATH=$PATH:$HADOOP_HOME/sbin