图文精华

About云-梭伦科技»专题 › 资源区 › 大数据资源 › Hive|数据仓库 › 项目源码› 大数据项目之电商数仓（用户行为数据采集）（三）
分享

大数据项目之电商数仓（用户行为数据采集）（三）

levycui 2020-11-25 20:22:28 发表于 项目源码 [显示全部楼层] 只看大图 回帖奖励 阅读模式 关闭右栏 0 3508
问题导读：
1、Flume有哪些组件？
2、日志采集如何配置Flume？
3、Flume的ETL和分类型拦截器有哪些？
4、日志采集Flume启动停止脚本如何设置？

上一篇：大数据项目之电商数仓（用户行为数据采集）（二）

4.4 采集日志Flume
2020-11-25_202140.jpg

4.4.1 日志采集Flume安装
集群规划：

2020-11-25_202551.jpg
4.4.2 项目经验之Flume组件
1）Source
（1）Taildir Source相比Exec Source、Spooling Directory Source的优势
TailDir Source：断点续传、多目录。Flume1.6以前需要自己自定义Source记录每次读取文件位置，实现断点续传。
Exec Source可以实时搜集数据，但是在Flume不运行或者Shell命令出错的情况下，数据将会丢失。
Spooling Directory Source监控目录，不支持断点续传。
（2）batchSize大小如何设置？
答：Event 1K左右时，500-1000合适（默认为100）
2）Channel
采用Kafka Channel，省去了Sink，提高了效率。

4.4.3 日志采集Flume配置
1）Flume配置分析

2020-11-25_202631.jpg

Flume直接读log日志的数据，log日志的格式是app-yyyy-mm-dd.log。
2）Flume的具体配置如下：
（1）在/opt/module/flume/conf目录下创建file-flume-kafka.conf文件
  1. [kgg@hadoop101 conf]$ vim file-flume-kafka.conf

  2. 在文件配置如下内容

  3. a1.sources=r1

  4. a1.channels=c1 c2

  5. 

  6. # configure source

  7. a1.sources.r1.type = TAILDIR

  8. a1.sources.r1.positionFile = /opt/module/flume/test/log_position.json

  9. a1.sources.r1.filegroups = f1

  10. a1.sources.r1.filegroups.f1 = /tmp/logs/app.+

  11. a1.sources.r1.fileHeader = true

  12. a1.sources.r1.channels = c1 c2

  13. 

  14. #interceptor

  15. a1.sources.r1.interceptors =  i1 i2

  16. a1.sources.r1.interceptors.i1.type = com.kgg.flume.interceptor.LogETLInterceptor$Builder

  17. a1.sources.r1.interceptors.i2.type = com.kgg.flume.interceptor.LogTypeInterceptor$Builder

  18. 

  19. a1.sources.r1.selector.type = multiplexing

  20. a1.sources.r1.selector.header = topic

  21. a1.sources.r1.selector.mapping.topic_start = c1

  22. a1.sources.r1.selector.mapping.topic_event = c2

  23. 

  24. # configure channel

  25. a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel

  26. a1.channels.c1.kafka.bootstrap.servers = hadoop101:9092,hadoop102:9092,hadoop103:9092

  27. a1.channels.c1.kafka.topic = topic_start

  28. a1.channels.c1.parseAsFlumeEvent = false

  29. a1.channels.c1.kafka.consumer.group.id = flume-consumer

  30. 

  31. a1.channels.c2.type = org.apache.flume.channel.kafka.KafkaChannel

  32. a1.channels.c2.kafka.bootstrap.servers = hadoop101:9092,hadoop102:9092,hadoop103:9092

  33. a1.channels.c2.kafka.topic = topic_event

  34. a1.channels.c2.parseAsFlumeEvent = false

  35. a1.channels.c2.kafka.consumer.group.id = flume-consumer
复制代码
注意：com.kgg.flume.interceptor.LogETLInterceptor和com.kgg.flume.interceptor.LogTypeInterceptor是自定义的拦截器的全类名。需要根据用户自定义的拦截器做相应修改。


2020-11-25_202710.jpg

4.4.4 Flume的ETL和分类型拦截器
本项目中自定义了两个拦截器，分别是：ETL拦截器、日志类型区分拦截器。
ETL拦截器主要用于，过滤时间戳不合法和Json数据不完整的日志

日志类型区分拦截器主要用于，将启动日志和事件日志区分开来，方便发往Kafka的不同Topic。
1）创建Maven工程flume-interceptor
2）创建包名：com.kgg.flume.interceptor
3）在pom.xml文件中添加如下配置
  1. <dependencies>

  2.     <dependency>

  3.         <groupId>org.apache.flume</groupId>

  4.         <artifactId>flume-ng-core</artifactId>

  5.         <version>1.7.0</version>

  6.     </dependency>

  7. </dependencies>

  8. 

  9. <build>

  10.     <plugins>

  11.         <plugin>

  12.             <artifactId>maven-compiler-plugin</artifactId>

  13.             <version>2.3.2</version>

  14.             <configuration>

  15.                 <source>1.8</source>

  16.                 <target>1.8</target>

  17.             </configuration>

  18.         </plugin>

  19.         <plugin>

  20.             <artifactId>maven-assembly-plugin</artifactId>

  21.             <configuration>

  22.                 <descriptorRefs>

  23.                     <descriptorRef>jar-with-dependencies</descriptorRef>

  24.                 </descriptorRefs>

  25.             </configuration>

  26.             <executions>

  27.                 <execution>

  28.                     <id>make-assembly</id>

  29.                     <phase>package</phase>

  30.                     <goals>

  31.                         <goal>single</goal>

  32.                     </goals>

  33.                 </execution>

  34.             </executions>

  35.         </plugin>

  36.     </plugins>

  37. </build>
复制代码

4）在com.kgg.flume.interceptor包下创建LogETLInterceptor类名
  1. Flume ETL拦截器LogETLInterceptor

  2. package com.kgg.flume.interceptor;

  3. 

  4. import org.apache.flume.Context;

  5. import org.apache.flume.Event;

  6. import org.apache.flume.interceptor.Interceptor;

  7. 

  8. import java.nio.charset.Charset;

  9. import java.util.ArrayList;

  10. import java.util.List;

  11. 

  12. public class LogETLInterceptor implements Interceptor {

  13. 

  14.     @Override

  15.     public void initialize() {

  16. 

  17.     }

  18. 

  19.     @Override

  20.     public Event intercept(Event event) {

  21. 

  22.         // 1 获取数据

  23.         byte[] body = event.getBody();

  24.         String log = new String(body, Charset.forName("UTF-8"));

  25. 

  26.         // 2 判断数据类型并向Header中赋值

  27.         if (log.contains("start")) {

  28.             if (LogUtils.validateStart(log)){

  29.                 return event;

  30.             }

  31.         }else {

  32.             if (LogUtils.validateEvent(log)){

  33.                 return event;

  34.             }

  35.         }

  36. 

  37.         // 3 返回校验结果

  38.         return null;

  39.     }

  40. 

  41.     @Override

  42.     public List<Event> intercept(List<Event> events) {

  43. 

  44.         ArrayList<Event> interceptors = new ArrayList<>();

  45. 

  46.         for (Event event : events) {

  47.             Event intercept1 = intercept(event);

  48. 

  49.             if (intercept1 != null){

  50.                 interceptors.add(intercept1);

  51.             }

  52.         }

  53. 

  54.         return interceptors;

  55.     }

  56. 

  57.     @Override

  58.     public void close() {

  59. 

  60.     }

  61. 

  62.     public static class Builder implements Interceptor.Builder{

  63. 

  64.         @Override

  65.         public Interceptor build() {

  66.             return new LogETLInterceptor();

  67.         }

  68. 

  69.         @Override

  70.         public void configure(Context context) {

  71. 

  72.         }

  73.     }

  74. }
复制代码

4）Flume日志过滤工具类
  1. package com.kgg.flume.interceptor;

  2. import org.apache.commons.lang.math.NumberUtils;

  3. 

  4. public class LogUtils {

  5. 

  6.     public static boolean validateEvent(String log) {

  7.         // 服务器时间 | json

  8.         // 1549696569054 | {"cm":{"ln":"-89.2","sv":"V2.0.4","os":"8.2.0","g":"M67B4QYU@gmail.com","nw":"4G","l":"en","vc":"18","hw":"1080*1920","ar":"MX","uid":"u8678","t":"1549679122062","la":"-27.4","md":"sumsung-12","vn":"1.1.3","ba":"Sumsung","sr":"Y"},"ap":"weather","et":[]}

  9. 

  10.         // 1 切割

  11.         String[] logContents = log.split("\\|");

  12. 

  13.         // 2 校验

  14.         if(logContents.length != 2){

  15.             return false;

  16.         }

  17. 

  18.         //3 校验服务器时间

  19.         if (logContents[0].length()!=13 || !NumberUtils.isDigits(logContents[0])){

  20.             return false;

  21.         }

  22. 

  23.         // 4 校验json

  24.         if (!logContents[1].trim().startsWith("{") || !logContents[1].trim().endsWith("}")){

  25.             return false;

  26.         }

  27. 

  28.         return true;

  29.     }

  30. 

  31.     public static boolean validateStart(String log) {

  32.  // {"action":"1","ar":"MX","ba":"HTC","detail":"542","en":"start","entry":"2","extend1":"","g":"S3HQ7LKM@gmail.com","hw":"640*960","l":"en","la":"-43.4","ln":"-98.3","loading_time":"10","md":"HTC-5","mid":"993","nw":"WIFI","open_ad_type":"1","os":"8.2.1","sr":"D","sv":"V2.9.0","t":"1559551922019","uid":"993","vc":"0","vn":"1.1.5"}

  33. 

  34.         if (log == null){

  35.             return false;

  36.         }

  37. 

  38.         // 校验json

  39.         if (!log.trim().startsWith("{") || !log.trim().endsWith("}")){

  40.             return false;

  41.         }

  42. 

  43.         return true;

  44.     }

  45. }
复制代码

5）Flume日志类型区分拦截器LogTypeInterceptor
  1. package com.kgg.flume.interceptor;

  2. 

  3. import org.apache.flume.Context;

  4. import org.apache.flume.Event;

  5. import org.apache.flume.interceptor.Interceptor;

  6. 

  7. import java.nio.charset.Charset;

  8. import java.util.ArrayList;

  9. import java.util.List;

  10. import java.util.Map;

  11. 

  12. public class LogTypeInterceptor implements Interceptor {

  13.     @Override

  14.     public void initialize() {

  15. 

  16.     }

  17. 

  18.     @Override

  19.     public Event intercept(Event event) {

  20. 

  21.         // 区分日志类型：   body  header

  22.         // 1 获取body数据

  23.         byte[] body = event.getBody();

  24.         String log = new String(body, Charset.forName("UTF-8"));

  25. 

  26.         // 2 获取header

  27.         Map<String, String> headers = event.getHeaders();

  28. 

  29.         // 3 判断数据类型并向Header中赋值

  30.         if (log.contains("start")) {

  31.             headers.put("topic","topic_start");

  32.         }else {

  33.             headers.put("topic","topic_event");

  34.         }

  35. 

  36.         return event;

  37.     }

  38. 

  39.     @Override

  40.     public List<Event> intercept(List<Event> events) {

  41. 

  42.         ArrayList<Event> interceptors = new ArrayList<>();

  43. 

  44.         for (Event event : events) {

  45.             Event intercept1 = intercept(event);

  46. 

  47.             interceptors.add(intercept1);

  48.         }

  49. 

  50.         return interceptors;

  51.     }

  52. 

  53.     @Override

  54.     public void close() {

  55. 

  56.     }

  57. 

  58.     public static class Builder implements  Interceptor.Builder{

  59. 

  60.         @Override

  61.         public Interceptor build() {

  62.             return new LogTypeInterceptor();

  63.         }

  64. 

  65.         @Override

  66.         public void configure(Context context) {

  67. 

  68.         }

  69.     }

  70. }
复制代码

6）打包
拦截器打包之后，只需要单独包，不需要将依赖的包上传。打包之后要放入Flume的lib文件夹下面。

2020-11-25_202746.jpg

注意：为什么不需要依赖包？因为依赖包在flume的lib目录下面已经存在了。
7）需要先将打好的包放入到hadoop101的/opt/module/flume/lib文件夹下面。
  1. [kgg@hadoop101 lib]$ ls | grep interceptor

  2. flume-interceptor-1.0-SNAPSHOT.jar
复制代码

8）分发Flume到hadoop102、hadoop103
  1. [kgg@hadoop101 module]$ xsync flume/

  2. [kgg@hadoop101 flume]$ bin/flume-ng agent --name a1 --conf-file conf/file-flume-kafka.conf &
复制代码

4.4.5 日志采集Flume启动停止脚本
1）在/home/kgg/bin目录下创建脚本f1.sh
  1. [kgg@hadoop101 bin]$ vim f1.sh

  2.     在脚本中填写如下内容

  3. #! /bin/bash

  4. 

  5. case $1 in

  6. "start"){

  7.         for i in hadoop101 hadoop102

  8.         do

  9.                 echo " --------启动 $i 采集flume-------"

  10.                 ssh $i "nohup /opt/module/flume/bin/flume-ng agent --conf-file /opt/module/flume/conf/file-flume-kafka.conf --name a1 -Dflume.root.logger=INFO,LOGFILE > /dev/null 2>&1 &"

  11.         done

  12. };;    

  13. "stop"){

  14.         for i in hadoop101 hadoop102

  15.         do

  16.                 echo " --------停止 $i 采集flume-------"

  17.                 ssh $i "ps -ef | grep file-flume-kafka | grep -v grep |awk '{print \$2}' | xargs kill"

  18.         done

  19. 

  20. };;

  21. esac
复制代码

说明1：nohup，该命令可以在你退出帐户/关闭终端之后继续运行相应的进程。nohup就是不挂起的意思，不挂断地运行命令。
说明2：/dev/null代表linux的空设备文件，所有往这个文件里面写入的内容都会丢失，俗称“黑洞”。
标准输入0：从键盘获得输入 /proc/self/fd/0
标准输出1：输出到屏幕（即控制台） /proc/self/fd/1
错误输出2：输出到屏幕（即控制台） /proc/self/fd/2
2）增加脚本执行权限
  1. [kgg@hadoop101 bin]$ chmod 777 f1.sh
复制代码

3）f1集群启动脚本
  1. [kgg@hadoop101 module]$ f1.sh start
复制代码

4）f1集群停止脚本
  1. [kgg@hadoop101 module]$ f1.sh stop
复制代码

最新经典文章，欢迎关注公众号



本帖被以下淘专辑推荐:

回复

使用道具 举报

自定义广告语

没找到任何评论，期待你打破沉寂

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
levycui

超级版主

关注

  • 654

    主题

  • 1167

    帖子

  • 97

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2025 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000