梅州站扩建,交换神器,软件网站设计,个人建网站要花多少钱[flume组件(原理上分析零点问题的由来)](#flume_15)[技术细节1](#1_27)[技术细节2](#2_77)[优化](#_200)[小结](#_219)概要
Flume解决零点漂移数据
例如#xff1a;
7号当天23:59:59的数据分类在7号,而不是因为数据传输的延迟分类在8号.
整体架构流程
javaflume
flume组… [flume组件(原理上分析零点问题的由来)](#flume_15)[技术细节1](#1_27)[技术细节2](#2_77)[优化](#_200)[小结](#_219)概要
Flume解决零点漂移数据
例如
7号当天23:59:59的数据分类在7号,而不是因为数据传输的延迟分类在8号.
整体架构流程
javaflume
flume组件(原理上分析零点问题的由来)
1.Flume由sourcechannelsink组成. 2.Flume数据传输的基本单元以Event的形式将数据从源头送至目的地。Event由Header和Body两部分组成. 图示: 解决:取Body的时间更换Header的timestamp (在source端加拦截器interceptor) 图示:
技术细节1
Flume配置文件:
#定义组件
a1.sourcesr1
a1.channelsc1
a1.sinksk1#配置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.topicstopic_log
a1.sources.r1.interceptors i1
a1.sources.r1.interceptors.i1.type com.cwl.database.flume.interceptor.TimestampInterceptor$Builder#配置channel
a1.channels.c1.type file
a1.channels.c1.checkpointDir /opt/module/flume/checkpoint/behavior1
a1.channels.c1.dataDirs /opt/module/flume/data/behavior1
a1.channels.c1.maxFileSize 2146435071
a1.channels.c1.capacity 1000000
a1.channels.c1.keep-alive 6#配置sink
a1.sinks.k1.type hdfs
a1.sinks.k1.hdfs.path /origin_data/database/log/topic_log/%Y-%m-%d
a1.sinks.k1.hdfs.filePrefix log
a1.sinks.k1.hdfs.round falsea1.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 gzip#组装
a1.sources.r1.channels c1
a1.sinks.k1.channel c1
注意配置中:(拦截器全类名) a1.sources.r1.interceptors.i1.type com.cwl.database.flume.interceptor.TimestampInterceptor$Builder
技术细节2
java编写拦截器代码: 依赖:
dependenciesdependencygroupIdorg.apache.flume/groupIdartifactIdflume-ng-core/artifactIdversion1.10.1/versionscopeprovided/scope/dependencydependencygroupIdcom.alibaba/groupIdartifactIdfastjson/artifactIdversion1.2.62/version/dependency
/dependenciesbuildpluginspluginartifactIdmaven-compiler-plugin/artifactIdversion2.3.2/versionconfigurationsource1.8/sourcetarget1.8/target/configuration/pluginpluginartifactIdmaven-assembly-plugin/artifactIdconfigurationdescriptorRefsdescriptorRefjar-with-dependencies/descriptorRef/descriptorRefs/configurationexecutionsexecutionidmake-assembly/idphasepackage/phasegoalsgoalsingle/goal/goals/execution/executions/plugin/plugins
/build
代码:
package com.cwl.database.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.Iterator;import java.util.List;
import java.util.Map;public class TimestampInterceptor implements Interceptor {Overridepublic void initialize() {}Overridepublic Event intercept(Event event) {//1、获取header和body的数据MapString, String headers event.getHeaders();String log new String(event.getBody(), StandardCharsets.UTF_8);try {//2、将body的数据类型转成jsonObject类型方便获取数据JSONObject jsonObject JSONObject.parseObject(log);//3、header中timestamp时间字段替换成日志生成的时间戳解决数据漂移问题String ts jsonObject.getString(ts);headers.put(timestamp, ts);return event;} catch (Exception e) {e.printStackTrace();return null;}
}Override
public ListEvent intercept(ListEvent list) {IteratorEvent iterator list.iterator();while (iterator.hasNext()) {Event event iterator.next();if (intercept(event) null) {iterator.remove();}}return list;
}**自我介绍一下小编13年上海交大毕业曾经在小公司待过也去过华为、OPPO等大厂18年进入阿里一直到现在。****深知大多数大数据工程师想要提升技能往往是自己摸索成长或者是报班学习但对于培训机构动则几千的学费着实压力不小。自己不成体系的自学效果低效又漫长而且极易碰到天花板技术停滞不前****因此收集整理了一份《2024年大数据全套学习资料》初衷也很简单就是希望能够帮助到想自学提升又不知道该从何学起的朋友。**
**既有适合小白学习的零基础资料也有适合3年以上经验的小伙伴深入学习提升的进阶课程基本涵盖了95%以上大数据开发知识点真正体系化****由于文件比较大这里只是将部分目录大纲截图出来每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频并且后续会持续更新****如果你觉得这些内容对你有帮助可以添加VXvip204888 备注大数据获取**
**一个人可以走的很快但一群人才能走的更远。不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人都欢迎扫码加入我们的的圈子技术交流、学习资源、职场吐槽、大厂内推、面试辅导让我们一起学习成长**笔记、源码讲义、实战项目、讲解视频并且后续会持续更新****如果你觉得这些内容对你有帮助可以添加VXvip204888 备注大数据获取**
[外链图片转存中...(img-YYa29ee3-1712962837543)]**一个人可以走的很快但一群人才能走的更远。不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人都欢迎扫码加入我们的的圈子技术交流、学习资源、职场吐槽、大厂内推、面试辅导让我们一起学习成长**
