1Flume配置概述

Flume需要将Kafka中topic_db主题的数据传输到HDFS故其需选用KafkaSource以及HDFSSinkChannel选用FileChannel。

HDFSSink需要将不同mysql业务表的数据写到不同的路径并且路径中应当包含一层日期用于区分每天的数据。

关键配置如下

2Flume配置实操

在hadoop104节点的Flume的job目录下创建kafka_to_hdfs_db.conf

[atguiguhadoop104

org.apache.flume.source.kafka.KafkaSource

5000

a1.sources.r1.batchDurationMillis

2000

a1.sources.r1.kafka.bootstrap.servers

topic_db

a1.sources.r1.kafka.consumer.group.id

flume

a1.sources.r1.interceptors.i1.type

com.atguigu.gmall.flume.interceptor.TimestampAndTableNameInterceptor$Buildera1.channels.c1.type

file

/opt/module/flume/checkpoint/behavior2

/opt/module/flume/data/behavior2/

2146435071

/origin_data/gmall/db/%{tableName}_inc/%Y-%m-%d

falsea1.sinks.k1.hdfs.rollInterval

134217728

dependenciesdependencygroupIdorg.apache.flume/groupIdartifactIdflume-ng-core/artifactIdversion1.9.0/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在com.atguigu.gmall.flume.interceptor包下创建TimestampAndTableNameInterceptor类

package

com.atguigu.gmall.flume.interceptor;

import

com.alibaba.fastjson.JSONObject;

import

org.apache.flume.interceptor.Interceptor;

import

java.nio.charset.StandardCharsets;

import

TimestampAndTableNameInterceptor

implements

StandardCharsets.UTF_8);JSONObject

jsonObject

JSONObject.parseObject(log);Long

jsonObject.getLong(ts);//Maxwell输出的数据中的ts字段时间戳单位为秒Flume

timeMills

jsonObject.getString(table);headers.put(timestamp,

timeMills);headers.put(tableName,

tableName);return

TimestampAndTableNameInterceptor

();}Overridepublic

将打好的包放入到hadoop104的/opt/module/flume/lib文件夹下

[atguiguhadoop102

flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar

1启动Zookeeper、Kafka集群

-Dflume.root.loggerinfo,console

3生成模拟数据

gmall2020-mock-db-2021-11-14.jar

若HDFS上的目标路径已有增量表的数据出现了就证明数据通道已经打通。

仔细观察会发现目标路径中的日期并非模拟数据的业务日期而是当前日期。

这是由于Maxwell输出的JSON字符串中的ts字段的值是数据的变动日期。

而真实场景下数据的业务日期与变动日期应当是一致的。

4编写Flume启停脚本

1在hadoop102节点的/home/atguigu/bin目录下创建脚本f3.sh

[atguiguhadoop102

/opt/module/flume/job/kafka_to_hdfs_db.conf

/dev/null

1Maxwell时间戳问题此处为了模拟真实环境对Maxwell源码进行了改动增加了一个参数mock_date该参数的作用就是指定Maxwell输出JSON字符串的ts时间戳的日期接下来进行测试。

修改Maxwell配置文件config.properties增加mock_date参数如下

kafka.bootstrap.servershadoop102:9092,hadoop103:9092#kafka

topic配置

kafka_topictopic_db#注该参数仅在maxwell教学版中存在修改该参数后重启Maxwell才可生效

mock_date2020-06-14#

jdbc_optionsuseSSLfalseserverTimezoneAsia/Shanghai

重启Maxwell

gmall2020-mock-db-2021-11-14.jar

2.2.6.4

通常情况下增量表需要在首日进行一次全量同步后续每日再进行增量同步首日全量同步可以使用Maxwell的bootstrap功能方便起见下面编写一个增量表首日全量同步脚本。

1在~/bin目录创建mysql_to_kafka_inc_init.sh

[atguiguhadoop102

该脚本的作用是初始化所有的增量表只需执行一次MAXWELL_HOME/opt/module/maxwellimport_data()

{$MAXWELL_HOME/bin/maxwell-bootstrap

--database

$MAXWELL_HOME/config.properties

}case

order_detail_activity)import_data

order_detail_coupon)import_data

order_info;;

order_detail_activityimport_data

user_info;;

2为mysql_to_kafka_inc_init.sh增加执行权限

[atguiguhadoop102

~/bin/mysql_to_kafka_inc_init.sh

3测试同步脚本

1在/home/atguigu/bin目录下创建脚本cluster.sh

[atguiguhadoop102