Flume1.9.0自定义拦截器

news/2024/11/16 22:40:21/

需求

1、在linux日志文件/data/log/moreInfoRes.log中一直会产生如下JSON数据：

{"id":"14943445328940974601","uid":"840717325115457536","lat":"53.530598","lnt":"-2.5620373","hots":0,"title":"0","status":"1","topicId":"0","end_time":"1494344570","watch_num":0,"share_num":"1","replay_url":null,"replay_num":0,"start_time":"1494344544","timestamp":1494344571,"type":"video_info"}
{"uid":"861848974414839801","nickname":"mick","usign":"","sex":1,"birthday":"","face":"","big_face":"","email":"abc@qq.com","mobile":"","reg_type":"102","last_login_time":"1494344580","reg_time":"1494344580","last_update_time":"1494344580","status":"5","is_verified":"0","verified_info":"","is_seller":"0","level":1,"exp":0,"anchor_level":0,"anchor_exp":0,"os":"android","timestamp":1494344580,"type":"user_info"}
{"send_id":"834688818270961664","good_id":"223","video_id":"14943443045138661356","gold":"10","timestamp":1494344574,"type":"gift_record"}

 2、需要根据数据中的type字段分目录存储，，并且还要对type字段的值进行一定的处理，最终处理之后的数据需要存储到HDFS上的/moreInfoRes目录中。例如：

  • ​type:video_info 类型的数据需要存储到 /moreInfoRes/videoInfo 目录里面。

  • ​type:user_info 类型的数据需要存储到 /moreInfoRes/userInfo 目录里面。

  • ​type:gift_record 类型的数据需要存储到 /moreInfoRes/giftRecord 目录里面。

3、这边拦截器用 Search and Replace Interceptor + Regex Extractor Interceptor 可以实现，但是这边使用前者的话效率有点低，故采用 自定义Interceptor + Regex Extractor Interceptor 实现。

实现

鉴于此，可以使用 Exec Source + Custom Interceptor + Regex Extractor Interceptor + File Channel + HDFS Sink 来实现。官方文档如下：

Exec Source:
https://flume.apache.org/releases/content/1.11.0/FlumeUserGuide.html#exec-sourceCustom Interceptor:
可参考其他 Interceptor 的实现Regex Extractor Interceptor:
https://flume.apache.org/releases/content/1.11.0/FlumeUserGuide.html#regex-extractor-interceptorFile Channel:
https://flume.apache.org/releases/content/1.11.0/FlumeUserGuide.html#file-channelHDFS Sink:
https://flume.apache.org/releases/content/1.11.0/FlumeUserGuide.html#hdfs-sink

创建工程

引入依赖

主要是 flume-ng-core 和 jackson 依赖，其他可不引入。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.example</groupId><artifactId>flume-demo</artifactId><version>1.0-SNAPSHOT</version><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><dependencies><dependency><groupId>org.apache.flume</groupId><artifactId>flume-ng-core</artifactId><version>1.9.0</version></dependency><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId><version>2.15.4</version></dependency>
<!--        <dependency>-->
<!--            <groupId>com.alibaba</groupId>-->
<!--            <artifactId>fastjson</artifactId>-->
<!--            <version>2.0.25</version>-->
<!--        </dependency>-->
<!--        <dependency>-->
<!--            <groupId>cn.hutool</groupId>-->
<!--            <artifactId>hutool-core</artifactId>-->
<!--            <version>5.8.27</version>-->
<!--        </dependency>--><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.10</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>1.7.10</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.33</version></dependency></dependencies></project>

编写 Custom Interceptor 

package com.example.flumedemo.interceptor;import com.example.flumedemo.constant.OptType;
import com.example.flumedemo.util.JsonUtil;
import com.example.flumedemo.util.NamingCaseUtil;
import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.apache.flume.interceptor.SearchAndReplaceInterceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.List;
import java.util.Map;/*** 自定义拦截器，* 将字段的值按照指定操作转换，得到的结果替换原来的值。** @author liaorj* @date 2024/11/13*/
public class MyInterceptor implements Interceptor {private static final Logger logger = LoggerFactory.getLogger(SearchAndReplaceInterceptor.class);/*** json中需要处理的字段*/private final String jsonField;/*** 需要对字段的值进行什么操作*/private final String optType;private final Charset charset;public MyInterceptor(String jsonField, String optType, Charset charset) {this.jsonField = jsonField;this.optType = optType;this.charset = charset;}@Overridepublic void initialize() {}@Overridepublic Event intercept(Event event) {try {logger.info("----event={}", JsonUtil.toJson(event));if (null == event || ArrayUtils.isEmpty(event.getBody())) {logger.info("----event or body is null");return event;}//将body转为map对象Map<String, Object> jsonObject;jsonObject = JsonUtil.toBean(event.getBody(), new TypeReference<Map<String, Object>>() {});logger.info("----jsonObject={}", jsonObject);//获取jsonField的值logger.info("----jsonField={}", this.jsonField);Object value = jsonObject.get(this.jsonField);logger.info("----jsonFieldValue={}", value);if (jsonObject.containsKey(this.jsonField)) {logger.info("----containsKey");} else {logger.info("----not containsKey");}logger.info("----jsonObject.keySet={}", jsonObject.keySet());if (jsonObject.keySet().contains(this.jsonField)) {logger.info("----keySet containsKey");} else {logger.info("----keySet not containsKey");}//如果含有下划线if (null != value) {String newValue = null;logger.info("-----opt={},code={}", this.optType, OptType.toCamelCase.getCode());logger.info("----opt equals={}", OptType.toCamelCase.getCode().equals(this.optType));if (OptType.toCamelCase.getCode().equals(this.optType)) {//将下划线字符串转为驼峰newValue = NamingCaseUtil.toCamelCase(value.toString());logger.info("----newValue={}", newValue);} else if (OptType.toKebabCase.getCode().equals(this.optType)) {//hutool和fastjson的类本地跑可以，上环境却用不了，执行到相关类就没有日志了，可能包冲突了，暂不用。
//                    newValue = NamingCase.toKebabCase(value.toString());} else if (OptType.toPascalCase.getCode().equals(this.optType)) {
//                    newValue = NamingCase.toPascalCase(value.toString());} else if (OptType.toUnderlineCase.getCode().equals(this.optType)) {newValue = NamingCaseUtil.toUnderlineCase2(value.toString());} else {newValue = value.toString();}//替换原来的值logger.info("----newValue2={}", newValue);jsonObject.put(this.jsonField, newValue);logger.info("----jsonObject2={}", jsonObject);event.setBody(JsonUtil.toJson(jsonObject).getBytes(charset));}} catch (Exception e) {throw new RuntimeException(e);}return event;}@Overridepublic List<Event> intercept(List<Event> events) {Iterator var2 = events.iterator();while (var2.hasNext()) {Event event = (Event) var2.next();this.intercept(event);}return events;}@Overridepublic void close() {}public static class Builder implements Interceptor.Builder {private String jsonField;private String optType;private Charset charset;public Builder() {this.charset = Charsets.UTF_8;}@Overridepublic void configure(Context context) {String jsonObjField = context.getString("jsonField");Preconditions.checkArgument(!StringUtils.isEmpty(jsonObjField), "Must supply a valid jsonField (may not be empty)");this.jsonField = jsonObjField;String optType = context.getString("optType");Preconditions.checkArgument(!StringUtils.isEmpty(optType), "Must supply a valid opt (may not be empty)");this.optType = optType;if (context.containsKey("charset")) {// May throw IllegalArgumentException for unsupported charsets.charset = Charset.forName(context.getString("charset"));}}@Overridepublic Interceptor build() {Preconditions.checkNotNull(this.jsonField, "jsonField required");Preconditions.checkNotNull(this.optType, "opt required");return new MyInterceptor(this.jsonField, this.optType, this.charset);}}/*public static void main(String[] args) {String str = "{\"send_id\":\"834688818270961664\",\"good_id\":\"223\",\"video_id\":\"14943443045138661356\",\"gold\":\"10\",\"timestamp\":1494344574,\"type\":\"gift_record\"}";MyInterceptor myInterceptor = new MyInterceptor("type", "toCamelCase", Charsets.UTF_8);Event event = new SimpleEvent();event.setBody(str.getBytes(StandardCharsets.UTF_8));Event result = myInterceptor.intercept(event);System.out.println(JsonUtil.toJson(JsonUtil.toBean(result.getBody(), new TypeReference<Map<String, Object>>() {})));}*/
}
package com.example.flumedemo.constant;/*** @author liaorj* @date 2024/11/13*/
public enum OptType {//将下划线方式命名的字符串转换为驼峰式。toCamelCase("toCamelCase"),//将驼峰式命名的字符串转换为短横连接方式。toKebabCase("toKebabCase"),//将下划线方式命名的字符串转换为帕斯卡式。toPascalCase("toPascalCase"),//将驼峰式命名的字符串转换为下划线方式toUnderlineCase("toUnderlineCase");private String code;OptType(String code) {this.code = code;}public String getCode() {return code;}
}

package com.example.flumedemo.util;import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ser.FilterProvider;
import com.fasterxml.jackson.databind.ser.impl.SimpleBeanPropertyFilter;
import com.fasterxml.jackson.databind.ser.impl.SimpleFilterProvider;import java.io.IOException;/*** @author liaorj* @date 2024/10/24*/
public class JsonUtil {private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();public static byte[] toBytes(Object object) {try {return OBJECT_MAPPER.writeValueAsBytes(object);} catch (JsonProcessingException e) {throw new RuntimeException(e);}}public static String toJson(Object object) {try {return OBJECT_MAPPER.writeValueAsString(object);} catch (JsonProcessingException e) {throw new RuntimeException(e);}}/*** 排除字段，敏感字段或太长的字段不显示：身份证、手机号、邮箱、密码等* 参考：https://www.baeldung-cn.com/jackson-ignore-properties-on-serialization** @param object* @param excludeProperties* @return*/public static String toJson(Object object, String[] excludeProperties) {try {SimpleBeanPropertyFilter theFilter = SimpleBeanPropertyFilter.serializeAllExcept(excludeProperties);FilterProvider filterProvider = new SimpleFilterProvider().addFilter("myFilter", theFilter);OBJECT_MAPPER.setFilterProvider(filterProvider);return OBJECT_MAPPER.writeValueAsString(object);} catch (JsonProcessingException e) {throw new RuntimeException(e);}}public static String toPrettyJson(Object object) {try {return OBJECT_MAPPER.writerWithDefaultPrettyPrinter().writeValueAsString(object);} catch (JsonProcessingException e) {throw new RuntimeException(e);}}public static <T> T toBean(String json, Class<T> clazz) {try {return OBJECT_MAPPER.readValue(json, clazz);} catch (JsonProcessingException e) {throw new RuntimeException(e);}}public static <T> T toBean(byte[] json, Class<T> clazz) {try {return OBJECT_MAPPER.readValue(json, clazz);} catch (IOException e) {throw new RuntimeException(e);}}public static <T> T toBean(byte[] json, TypeReference<T> typeReference) {try {return OBJECT_MAPPER.readValue(json, typeReference);} catch (IOException e) {throw new RuntimeException(e);}}
}

package com.example.flumedemo.util;import java.util.regex.Matcher;
import java.util.regex.Pattern;/*** 字符串转化工具类* @author liaorj* @date 2024/11/14*/
public class NamingCaseUtil {private static Pattern linePattern = Pattern.compile("_(\\w)");/*** 下划线转驼峰*/public static String toCamelCase(String str) {str = str.toLowerCase();Matcher matcher = linePattern.matcher(str);StringBuffer sb = new StringBuffer();while (matcher.find()) {matcher.appendReplacement(sb, matcher.group(1).toUpperCase());}matcher.appendTail(sb);return sb.toString();}/*** 驼峰转下划线(简单写法，效率低于{@link #toUnderlineCase2(String)})*/public static String toUnderlineCase(String str) {return str.replaceAll("[A-Z]", "_$0").toLowerCase();}private static Pattern humpPattern = Pattern.compile("[A-Z]");/*** 驼峰转下划线,效率比上面高*/public static String toUnderlineCase2(String str) {Matcher matcher = humpPattern.matcher(str);StringBuffer sb = new StringBuffer();while (matcher.find()) {matcher.appendReplacement(sb, "_" + matcher.group(0).toLowerCase());}matcher.appendTail(sb);return sb.toString();}
}

打包

打包前可以使用MyInterceptor类的main函数测试下。
mvn clean
mvn package
打包好后，需要把当前jar包上传到linux上的flume目录下的lib目录中。

配置文件

然后在flume目录下的conf目录下创建配置文件：file-to-hdfs-customInterceptor.conf，内容如下，注意自定义拦截器所在包和HDFS主机ip要修改成自己的。

# example.conf: A single-node Flume configuration# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /data/log/moreInfoRes.log# Describe/configure the source interceptors
a1.sources.r1.interceptors = i1 i2 
a1.sources.r1.interceptors.i1.type = com.example.flumedemo.interceptor.MyInterceptor$Builder
a1.sources.r1.interceptors.i1.jsonField = type
a1.sources.r1.interceptors.i1.optType = toCamelCasea1.sources.r1.interceptors.i2.type = regex_extractor
a1.sources.r1.interceptors.i2.regex = "type":"(\\w+)"
a1.sources.r1.interceptors.i2.serializers = s1
a1.sources.r1.interceptors.i2.serializers.s1.name = logType# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://192.168.163.130:9000/moreInfoRes/%{logType}
a1.sinks.k1.hdfs.filePrefix = data
a1.sinks.k1.hdfs.fileSuffix = .log
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.writeFormat = Text
a1.sinks.k1.hdfs.rollInterval = 3600
a1.sinks.k1.hdfs.rollSize = 134217728
a1.sinks.k1.hdfs.rollCount = 0
a1.sinks.k1.hdfs.useLocalTimeStamp = true# Use a channel which buffers events in memory
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /data/moreInfoRes/checkpointDir
a1.channels.c1.dataDirs = /data/moreInfoRes/dataDirs# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

启动flume

切换到flume目录，执行：

bin/flume-ng agent --name a1 --conf conf --conf-file conf/file-to-hdfs-customInterceptor.conf -Dflume.root.logger=INFO,console

测试结果

执行 hdfs dfs -ls -R / 命令查看 HDFS上 的 /moreInfoRes 目录文件信息，可以看到处理成功了：

 


http://www.ppmy.cn/news/1547558.html

相关文章

服务器能支持承载多少访问量？？

服务器能支持承载多少访问量？？

服务器能支持承载的访问量是一个受多种因素影响的复杂问题。 以下是对该问题的详细分析&#xff1a; 一、关键影响因素 服务器的硬件配置&#xff1a; 处理器性能&#xff1a;服务器的CPU性能决定了其处理请求的速度和能力。高性能的CPU能够更快速地处理并发请求&#xff0c;…
阅读更多...
提升治超站效率：因泰立科技激光式车辆长宽高检测系统

提升治超站效率：因泰立科技激光式车辆长宽高检测系统

ILS-04激光式车辆长宽高检测系统是因泰立科技自主研发的一套全自动车辆长宽高检测系统。已广泛地应用于固定式治超站、高速公路入口治超站、非现场执法站、物料场、停车场的车辆长宽高检测。 激光车辆检测技术的优点&#xff1a;检测精度明显高于目前同类的非接触检测技术如微波…
阅读更多...
C/C++语言基础--initializer_list表达式、tuple元组、pair对组简介

C/C++语言基础--initializer_list表达式、tuple元组、pair对组简介

本专栏目的 更新C/C的基础语法&#xff0c;包括C的一些新特性 前言 initializer_list表达式、tuple元组、pair对组再C日常还是比较常用的&#xff0c;尤其是对组在刷算法还是挺好用的&#xff0c;这里做一个简介&#xff1b;这三个语法结合C17的结构化绑定会更好用&#xff…
阅读更多...
创客节初中组C++模拟题

创客节初中组C++模拟题

来源：加码未来2024年深圳罗湖区创客节模拟题（初中组） 第一题 题目描述 通常为了节约时间，我们会用一个字母代替一个单词，比如Y代替Yes，N代替No。给出一个字母，输出其代替的单词 输入格式 一个字母，可能为大写或小写，但只会是yYnN之一。 输出格式 Yes 或 No 样例输…
阅读更多...
HTML（超文本标记语言）

HTML（超文本标记语言）

HTML&#xff08;超文本标记语言 - HyperText Markup Language&#xff09;是一种用于创建网页的标准标记语言。 HTML 最初是由蒂姆・伯纳斯 - 李&#xff08;Tim Berners - Lee&#xff09;在 1990 年左右开发的。当时的目的是为了让世界各地的科学家能够方便地共享和交流信息…
阅读更多...
深度学习每周学习总结J5（DenseNet-121 +SE 算法实战与解析 - 猴痘识别）

深度学习每周学习总结J5（DenseNet-121 +SE 算法实战与解析 - 猴痘识别）

&#x1f368; 本文为&#x1f517;365天深度学习训练营 中的学习记录博客&#x1f356; 原作者&#xff1a;K同学啊 | 接辅导、项目定制 0. 总结 数据导入及处理部分&#xff1a;本次数据导入没有使用torchvision自带的数据集&#xff0c;需要将原始数据进行处理包括数据导入…
阅读更多...
CentOS 修改服务器登录密码的完整指南

CentOS 修改服务器登录密码的完整指南

个人名片 &#x1f393;作者简介&#xff1a;java领域优质创作者 &#x1f310;个人主页&#xff1a;码农阿豪 &#x1f4de;工作室&#xff1a;新空间代码工作室&#xff08;提供各种软件服务&#xff09; &#x1f48c;个人邮箱&#xff1a;[2435024119qq.com] &#x1f4f1…
阅读更多...
stm32 spi读写W25Q128实例

stm32 spi读写W25Q128实例

文章目录 一、W25Q128芯片简介二、SPI初始化与配置三、W25Q128命令帧格式与操作写使能&#xff08;0x06&#xff09;&#xff1a;读取状态寄存器&#xff08;0x05&#xff09;&#xff1a;读取数据&#xff08;0x03&#xff09;&#xff1a;页编程&#xff08;0x02&#xff09;…
阅读更多...
最新文章

Copyright @ 2022~2023