文章目录
8. 实时数仓同步数据
实时数仓由Flink源源不断从Kafka当中读数据计算，所以不需要手动同步数据到实时数仓。
9. 离线数仓同步数据
9.1 用户行为数据同步
9.1.1 数据通道
用户行为数据由Flume从Kafka直接同步到HDFS，由于离线数仓采用Hive的分区表按天统计，所以目标路径要包含一层日期。具体数据流向如下图所示。
9.1.1.1 用户行为数据通道
9.1.2 日志消费Flume配置概述
按照规划，该Flume需将Kafka中topic_log的数据发往HDFS。并且对每天产生的用户行为日志进行区分，将不同天的数据发往HDFS不同天的路径。
此处选择KafkaSource、FileChannel、HDFSSink。
关键配置如下：
9.1.2.1 日志消费Flume关键配置
9.1.3 日志消费Flume配置实操
9.1.3.1 创建Flume配置文件
先将里面的文件删除。
9.1.3.2 配置文件内容如下
# 定义组件
a1.sources=r1
a1.channels=c1
a1.sinks=k1
# 配置source1
# 连接Kafka集群
a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.r1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092
# 消费者从topic_log读数据
a1.sources.r1.kafka.topics = topic_log
# 给消费者设置一个唯一标识，也就是消费者组ID，建议配置和业务一样的，如果用默认的，有多个消费者，则会有的消费者可以消费到数据，有些消费者消费不了数据
a1.sources.r1.kafka.consumer.group.id = topic_log
# 一次最大传输多少数据，数据满了就传输
a1.sources.r1.batchSize = 2000
# 多久传输一次，时间到了就传输，不管数据到2000了没有，时间一到就开始传输数据。
a1.sources.r1.batchDurationMillis = 1000
# 拦截器
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.summer.gmall.flume.interceptor.TimestampInterceptor$Builder
# 配置channel
a1.channels.c1.type = file
# 这个是Flume的第一次备份路径
a1.channels.c1.checkpointDir = /opt/module/flume-1.9.0/checkpoint/behavior1
# 打开二次Flume的二次备份，一般情况这个都是关闭状态，要是想打开的话改为true即可，然后配置backupCheckpointDir 的路径，这个路径要和第一次的路径放在不同的磁盘上
a1.channels.c1.useDualCheckpoints = false
# 这个是Flume的第二次备份路径
# a1.channels.c1.backupCheckpointDir = /opt/module/flume-1.9.0/checkpoint/behavior2
# 是flume的多目录存储，可以将服务器存在多个磁盘上，目前我只有一个磁盘，因此只能配置一个磁盘。和Hadoop的datadirs是一个意思。
a1.channels.c1.dataDirs = /opt/module/flume-1.9.0/data/behavior1
# file channel写入文件，这个文件最大可以容纳2G内容
a1.channels.c1.maxFileSize = 2146435071
# 是file channel条数的限制，最大是1000000
a1.channels.c1.capacity = 1000000
# 等一会，如果还没有足够的空间，还是会回滚
a1.channels.c1.keep-alive = 3
# 配置sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /origin_data/gmall/log/topic_log/%Y-%m-%d
a1.sinks.k1.hdfs.filePrefix = log
# 如果想配置6个小时落一次盘的话，可以将round打开，将hdfs.roundValue这个参数设置为6，hdfs.roundUnit这个参数设置为hour
# 并且a1.sinks.k1.hdfs.path = /origin_data/gmall/log/topic_log/%Y-%m-%d/%h要多加一个%h
a1.sinks.k1.hdfs.round = false
# 解决数据漂移问题
# 多少秒后滚动一次
a1.sinks.k1.hdfs.rollInterval = 10
# 文件大小超过设置值后滚动一次
a1.sinks.k1.hdfs.rollSize = 134217728
# event的条数，当设置多少条后，就是多少条后滚动一次
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
| hdfs.round | false | Should the timestamp be rounded down (if true, affects all time based escape sequences except %t) |
| hdfs.roundValue | 1 | Rounded down to the highest multiple of this (in the unit configured using hdfs.roundUnit), less than current time. |
| hdfs.roundUnit | second | The unit of the round down value – second, minute or hour. |
9.1.3.2.1 配置优化
9.1.3.2.1.1 FileChannel优化
通过配置dataDirs指向多个路径，每个路径对应不同的硬盘，增大Flume吞吐量。
官方说明如下：
Comma separated list of directories for storing log files. Using multiple directories on separate disks can improve file channel peformance
checkpointDir和backupCheckpointDir也尽量配置在不同硬盘对应的目录中，保证checkpoint坏掉后，可以快速使用backupCheckpointDir恢复数据
9.1.3.2.1.2 HDFS Sink优化
1.HDFS存入大量小文件，有什么影响？
元数据层面：每个小文件都有一份元数据，其中包括文件路径，文件名，所有者，所属组，权限，创建时间等，这些信息都保存在Namenode内存中。所以小文件过多，会占用Namenode服务器大量内存，影响Namenode性能和使用寿命
计算层面：默认情况下MR会对每个小文件启用一个Map任务计算，非常影响计算性能。同时也影响磁盘寻址时间。
2.HDFS小文件处理
官方默认的这三个参数配置写入HDFS后会产生小文件，hdfs.rollInterval、hdfs.rollSize、hdfs.rollCount
基于以上hdfs.rollInterval=3600，hdfs.rollSize=134217728，hdfs.rollCount =0几个参数综合作用，效果如下：
（1）文件在达到128M时会滚动生成新文件
（2）文件创建超3600秒时会滚动生成新文件
9.1.3.3 编写Flume拦截器
9.1.3.3.1 数据漂移问题
9.1.3.3.2 在com.summer.gmall.flume.interceptor包下创建TimestampInterceptor类
package com.summer.gmall.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.List;
import java.util.Map;
/**
* @author Redamancy
* @create 2022-10-29 19:54
*/
public class TimestampInterceptor implements Interceptor {
@Override
public void initialize() {
}
@Override
public Event intercept(Event event) {
//1 获取header和body的数据
Map<String, String> headers = event.getHeaders();
byte[] body = event.getBody();
String log = new String(body, StandardCharsets.UTF_8);
//2 将body的数据类型转成jsonObject类型（方便获取数据）
JSONObject jsonObject = JSONObject.parseObject(log);
//3 header中timestamp时间字段替换成日志生产的时间戳（解决数据漂移问题）
String ts = jsonObject.getString("ts");
headers.put("timestamp", ts);
return event;
}
@Override
public List<Event> intercept(List<Event> list) {
for (Event event : list) {
intercept(event);
}
return list;
}
@Override
public void close() {
}
public static class Builder implements Interceptor.Builder {
@Override
public Interceptor build() {
return new TimestampInterceptor();
}
@Override
public void configure(Context context) {
}
}
}
9.1.3.3.3 重新打包
先clean再package
9.1.3.3.4 需要先将打好的包放入到hadoop104的/opt/module/flume-1.9.0/lib文件夹下面
需要先在这个目录下面看看有没有这个名字的jar包，如果有则先删除rm，然后再将写的jar上传到这个目录下面，如果不先删除该jar包，则上传的jar名字后面会加个.0，jar包也不会运行起来。
9.1.4 日志消费Flume测试
9.1.4.1 启动Zookeeper、Kafka集群
9.1.4.2 启动日志采集Flume
[summer@hadoop102 module]$ f1.sh start
9.1.4.3 启动hadoop104的日志消费Flume
[summer@hadoop104 flume-1.9.0]$ bin/flume-ng agent -n a1 -c conf/ -f job/kafka_to_hdfs_log.conf -Dflume.root.logger=info,console
9.1.4.4 生成模拟数据
[summer@hadoop102 ~]$ lg.sh
9.1.4.5 观察HDFS是否出现数据
生成模拟数据之前
生成模拟数据之后
9.1.5 日志消费Flume启停脚本
若上述测试通过，为方便，此处创建一个Flume的启停脚本。
9.1.5.1 在hadoop102节点的/home/summer/bin目录下创建脚本f2.sh
[summer@hadoop102 bin]$ vim f2.sh
在脚本中填写如下内容
#!/bin/bash
case $1 in
"start")
echo " --------启动 hadoop104 下游flume-------"
ssh hadoop104 "nohup /opt/module/flume-1.9.0/bin/flume-ng agent -n a1 -c /opt/module/flume-1.9.0/conf -f /opt/module/flume-1.9.0/job/kafka_to_hdfs_log.conf >/dev/null 2>&1 &"
;;
"stop")
echo " --------停止 hadoop104 下游flume-------"
ssh hadoop104 "ps -ef | grep kafka_to_hdfs_log | grep -v grep |awk '{print \$2}' | xargs -n1 kill"
;;
esac
9.1.5.2 增加脚本执行权限
[summer@hadoop102 bin]$ chmod 777 f2.sh
9.1.5.3 f2启动
[summer@hadoop102 bin]$ f2.sh start
9.1.5.4 f2停止
[summer@hadoop102 bin]$ f2.sh stop
文章出处登录后可见！
已经登录？立即刷新
