大数据-225 离线数仓 - 目前需求分析指标口径日志数据采集 taildir source HDFS Sink Agent Flume 优化配置

2024-11-16 19:19:45

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

Hadoop（已更完）
HDFS（已更完）
MapReduce（已更完）
Hive（已更完）
Flume（已更完）
Sqoop（已更完）
Zookeeper（已更完）
HBase（已更完）
Redis （已更完）
Kafka（已更完）
Spark（已更完）
Flink（已更完）
ClickHouse（已更完）
Kudu（已更完）
Druid（已更完）
Kylin（已更完）
Elasticsearch（已更完）
DataX（已更完）
Tez（已更完）
数据挖掘（已更完）
Prometheus（已更完）
Grafana（已更完）
离线数仓（正在更新…）

章节内容

上节我们完成了如下的内容：

技术选型
版本选型
系统逻辑架构
数据库命名规范

需求分析

会员数据是后期营销的很重要的数据，网店会专门针对会员进行一系列营销活动，电商会员一般门槛较低，注册网站即可加入，有些电商平台的高级会员具有时效性，需要购买的VIP会员卡或一年内消费达到多少才能成为高级会员。
计算指标
新增会员：每次新增会员数
活跃会员：每日、每周、每月的活跃会员数
会员留存：1、2、3日会员的留存数，1、2、3日的会员留存率

指标口径业务逻辑

会员：以设备为判断标准，每个独立设备认为是一个会员，Android系统通常根据IMEI号，IOS系统通常根据OpenUDID来标识一个独立会员，每部移动设备是一个会员。
活跃会员：打开应用的会员即为活跃会员，暂不考虑用户的实际使用情况，一台设备每天多次打开计算为一个活跃会员，在自然周内启动过应用的会员为周活跃会员，同理还有月活跃会员。
会员活跃率：一天内活跃会员数与总会员数的比率是日活跃率，还有周活跃率（自然周）、月活跃率（自然月）
新增会员：第一次使用应用的用户，定义为新增会员，卸载再次安装的设备，不会被算作一次新增。新增用户包括日新增会员、周（自然周）新增会员、月（自然月）新增会员
留存会员与留存率：某段时间的新增会员，经过一段时间后，仍继续使用应用认为是留存会员，这部分会员占当时新增会员的比例为留存率。

已知条件是：

明确了需求
输入：启动日志、事件日志
输出：新增会员、活跃会员、留存会员
日志文件：ODS、DWD、DWS、ADS（输出）

下一步做什么？
数据采集：日志文件到 Flume 到 HDFS 到 ODS

日志数据采集

原始日志数据（一条启动日志）

2020-07-30 14:18:47.339 [main] INFO com.ecommerce.AppStart - {"app_active":{"name":"app_active","json":{"entry":"1","action":"1","error_code":"0"},"time":1596111888529},"attr":{"area":"泰安","uid":"2F10092A9","app_v":"1.1.13","event_type":"common","device_id":"1FB872-9A1009","os_type":"4.7.3","channel":"DK","language":"chinese","brand":"iphone-9"}}

数据采集的流程：

选择Flume作为采集日志数据的工具：

Flume1.6 无论是 Spooling Directory Source、Exec Source均不能很好的满足动态实时收集的需求
Flume1.8+提供了一个非常好的Taildir Source，使用该Source可以监控多个目录，对目录中写入的数据进行实时采集。

taildir source

taildir source特点

使用正则表达式匹配目录中的文件名
监控的文件中，一旦有数据写入，Flume就会将信息写入到指定的Sink
高可靠，不会丢失数据
不会对跟踪文件有任何处理，不会重命名也不会删除
不支持Windows，不能读二进制文件，支持按行读取文本文件

tail source配置

positionFile：配置检查点文件的路径，检查点文件会以JSON格式保存已经读取文件的位置，解决断点续传的问题
filegroups：指定filegroups，可以有多个，以空格分隔（taildir source可以同时监控多个目录中的文件）
filegroups.f1：配置每个filegroup的文件绝对路径，文件名可以用正则表达式匹配

HDFS Sink配置

HDFS Sink都会采用滚动生成文件的方式，滚动生成文件的策略有：

基于时间，hdfs.rollInterval 30秒
基于文件大小，hdfs.rollSize 1024字节
基于Event数量，hdfs.rollCount 10个event
基于文件空闲时间，hdfs.idleTimeout 0
0 禁用
minBlockReplicase，默认值与HDFS副本数一致。设为1是为了让Flume感知不到HDFS的块复制，此时其他的滚动方式配置（时间间隔、文件大小、events数量）才不会收到影响

Agent配置

我把配置放在了这里：

cd /opt/wzk
cd flume-conf
vim flume-log2hdfs1.conf

配置的内容如下所示：

a1.sources = r1
a1.sinks = k1
a1.channels = c1

# taildir source
a1.sources.r1.type = TAILDIR
a1.sources.r1.positionFile = /opt/wzk/conf/startlog_position.json
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 = /opt/wzk/logs/start/.*log

# memorychannel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 100000
a1.channels.c1.transactionCapacity = 2000

# hdfs sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /user/data/logs/start/%Y-%m-%d/
a1.sinks.k1.hdfs.filePrefix = startlog.
a1.sinks.k1.hdfs.fileType = DataStream
# 配置文件滚动方式（文件大小32M）
a1.sinks.k1.hdfs.rollSize = 33554432
a1.sinks.k1.hdfs.rollCount = 0
a1.sinks.k1.hdfs.rollInterval = 0
a1.sinks.k1.hdfs.idleTimeout = 0
a1.sinks.k1.hdfs.minBlockReplicas = 1
# 向hdfs上刷新的event的个数
a1.sinks.k1.hdfs.batchSize = 1000
# 使用本地时间
a1.sinks.k1.hdfs.useLocalTimeStamp = true

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

对应的截图如下所示：

Flume的优化配置

使用如下的指令，启动Agent进行测试：

flume-ng agent --conf-file /opt/wzk/flume-conf/flume-log2hdfs1.conf -name a1 -Dflum
e.roog.logger=INFO,console

启动后的截图如下所示：

查看刚才的Flume窗口：

查看HDFS的内容：

码农公寓