Flume部署总结

Posted 2021-02-02 yanxun1

 

拦截器

一、需求的产生

因生产需求，要将kafka中的数据上传至hdfs，所以计划部署flume来完成。

? 1.需要将kafkaSource中的json数据解析成需要的数据格式，落地至hdfs，供hive加载，所以此处需要自定义拦截器，对event.body进行逻辑解析。

? 2.同时因为不同kafkaTopic消息需要落地至hdfs不同路径，需要对event进行加头处理。此处会按照uid和从数据解析获得的time，对key为“timestamp” 的头进行设置。

二、架构设计

由于需要收集kafkaTopic数量较多，防止flume挂掉，故布置两层flume：

? （1）第一层：kafkaSource-fileChannel-avroSink

? 为保证数据安全故采用fileChannel，为了容灾，配置sinkGroup，类型为failover。

? （2）第二层：avroSource-fileChannel-hdfsSink

? 因为需要sink到不同路径中，所以添加挑选器，type = multiplexing 根据自定义的头选择channel，不同 的channel对应不同的通到。拦截器为自定义拦截器。sink则可以根据拦截器设置timestampe，生成%Y%m%d来生成hdfs路径。

三、开发中遇到的问题

1）多路复用的多线程修改

由于需要对event加头，所以初始化一个hashmap Headers来装入header和value，最初为了节约内存空间，就生命了一个全局变量，但是测试就发现，数据错乱，进入了它不应该进入的路径及文件。猜测是因为对线程调用xxx方法。对Headers进行修改。因此改为在方法内声明变量，问题就解决了。

注：单个sink的情况下不会出现。

四、收获

测试过程中需要自己生产数据，灌入kafka Topic中。

1)动态获取配置文件的路径

代码中的一些变量，想要动态的从配置文件中获取，而配置文件得位置又不确定，因此，只要放在jar包同级目录的conf下就好了，至于父级目录由代码动态获取就好了。而获取父级目录的逻辑又和系统相关。

LInux代码：

String rootPath = GetJarFatherPath.class.getProtectionDomain().getCodeSource().getLocation().getPath();
String fatherPath = rootPath.substring(0,rootPath.lastIndexOf("/"));
String basePath = fatherPath + "/conf/";
        try {
            FileInputStream in = new FileInputStream(basePath + "配置文件名字");
            Properties pro = new Properties();
            pro.load(in);
            in.close();
        } catch (Exception e) {
            e.printStackTrace();
        } 

2)多线程执行任务后自动结束

for(int i = n;i < n; i++){
  
    pool.execute(new Runnable)
    
}
?
pool.shutdown();//  完成在执行的任务，不再接受新任务

补充：线程池几种状态：

1、RUNNING

(1) 状态说明：线程池处在RUNNING状态时，能够接收新任务，以及对已添加的任务进行处理。 (02) 状态切换：线程池的初始化状态是RUNNING。换句话说，线程池被一旦被创建，就处于RUNNING状态，并且线程池中的任务数为0！

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

2、 SHUTDOWN

(1) 状态说明：线程池处在SHUTDOWN状态时，不接收新任务，但能处理已添加的任务。 (2) 状态切换：调用线程池的shutdown()接口时，线程池由RUNNING -> SHUTDOWN。

3、STOP

(1) 状态说明：线程池处在STOP状态时，不接收新任务，不处理已添加的任务，并且会中断正在处理的任务。 (2) 状态切换：调用线程池的shutdownNow()接口时，线程池由(RUNNING or SHUTDOWN ) -> STOP。

4、TIDYING

(1) 状态说明：当所有的任务已终止，ctl记录的”任务数量”为0，线程池会变为TIDYING状态。当线程池变为TIDYING状态时，会执行钩子函数terminated()。terminated()在ThreadPoolExecutor类中是空的，若用户想在线程池变为TIDYING时，进行相应的处理；可以通过重载terminated()函数来实现。 (2) 状态切换：当线程池在SHUTDOWN状态下，阻塞队列为空并且线程池中执行的任务也为空时，就会由 SHUTDOWN -> TIDYING。 当线程池在STOP状态下，线程池中执行的任务为空时，就会由STOP -> TIDYING。

5、 TERMINATED

(1) 状态说明：线程池彻底终止，就变成TERMINATED状态。 (2) 状态切换：线程池处在TIDYING状态时，执行完terminated()之后，就会由 TIDYING -> TERMINATED。