旅游网站建设方案背景描述,万维网的代表网站,环境设计,wordpress修改地址后网站打不开业务数据采集与同步 业务采集组件配置业务数据同步概述数据同步策略选择数据同步工具概述1.1.4 全量表数据同步DataX配置文件生成全量表数据同步脚本增量表数据同步 MySQL - Maxwell - Kafka - Flume - HDFSMaxwell配置增量表首日全量同步 业务采集组件配置 Maxwell将业务采集到… 业务数据采集与同步 业务采集组件配置业务数据同步概述数据同步策略选择数据同步工具概述1.1.4 全量表数据同步DataX配置文件生成全量表数据同步脚本增量表数据同步 MySQL - Maxwell - Kafka - Flume - HDFSMaxwell配置增量表首日全量同步 业务采集组件配置 Maxwell将业务采集到的kafka下游从kafka消费数据用于计算 链接: 配置Maxwell
业务数据同步概述
业务数据是数据仓库的重要数据来源我们需要每日定时从业务数据库中抽取数据传输到数据仓库中之后再对数据进行分析统计。为保证统计结果的正确性需要保证数据仓库中的数据与业务数据库是同步的离线数仓的计算周期通常为天所以数据同步周期也通常为天即每天同步一次即可。数据的同步策略有全量同步和增量同步。 全量同步就是每天都将业务数据库中的全部数据同步一份到数据仓库这是保证两侧数据同步的最简单的方式。 增量同步就是每天只将业务数据中的新增及变化数据同步到数据仓库。采用每日增量同步的表通常需要在首日先进行一次全量同步。
数据同步策略选择
两种策略都能保证数据仓库和业务数据库的数据同步那应该如何选择呢下面对两种策略进行简要对比。 若业务表数据量比较大且每天数据变化的比例比较低这时应采用增量同步否则可采用全量同步。
数据同步工具概述
数据同步工具种类繁多大致可分为两类 一类是以DataX、Sqoop为代表的基于Select查询的离线、批量同步工具 另一类是以Maxwell、Canal、Flink-CDC为代表的基于数据库数据变更日志例如MySQL的binlog其会实时记录所有的insert、update以及delete操作的实时流式同步工具。 全量同步通常使用DataX、Sqoop等基于查询的离线同步工具。而增量同步既可以使用DataX、Sqoop等工具也可使用Maxwell、Canal、Flink-CDC等工具下面对增量同步不同方案进行简要对比。
1.1.4 全量表数据同步
链接: 配置DataX 数据通道 全量表数据由DataX从MySQL业务数据库直接同步到HDFS具体数据流向如下图所示。 DataX配置文件 我们需要为每张全量表编写一个DataX的json配置文件此处以 user为例配置文件内容如下
{job: {content: [{reader: {name: mysqlreader,parameter: {column: [id,create_time,update_time,email,hashed_password,telephone,username],connection: [{jdbcUrl: [jdbc:mysql://hadoop102:3306/medical?useSSLfalseallowPublicKeyRetrievaltrueuseUnicodetruecharacterEncodingutf-8],table: [user]}],password: 000000,splitPk: ,username: root}},writer: {name: hdfswriter,parameter: {column: [{name: id,type: bigint},{name: create_time,type: string},{name: update_time,type: string},{name: email,type: string},{name: hashed_password,type: string},{name: telephone,type: string},{name: username,type: string}],compress: gzip,defaultFS: hdfs://hadoop102:8020,fieldDelimiter: \t,fileName: user,fileType: text,path: ${targetdir},writeMode: truncate,nullFormat: }}}],setting: {speed: {channel: 1}}}
}由于目标路径包含一层日期用于对不同天的数据加以区分故path参数并未写死需在提交任务时通过参数动态传入参数名称为targetdir。
DataX配置文件生成
链接: DataX配置文件生成 将生成器上传到服务器的/opt/module/gen_datax_config目录 [atguiguhadoop102 ~]$ mkdir /opt/module/gen_datax_config [atguiguhadoop102 ~]$ cd /opt/module/gen_datax_config 上传生成器 修改configuration.properties配置
mysql.usernameroot
mysql.password000000
mysql.hosthadoop102
mysql.port3306
mysql.database.importmedical
# mysql.database.export
mysql.tables.importdict,doctor,hospital,medicine,patient,user
# mysql.tables.export
is.seperated.tables0
hdfs.urihdfs://hadoop102:8020
import_out_dir/opt/module/datax/job/medical/import
# export_out_dir执行
[atguiguhadoop102 gen_datax_config]$ java -jar datax-config-generator-1.0-SNAPSHOT-jar-with-dependencies.jar测试生成的DataX配置文件 以user为例测试用脚本生成的配置文件是否可用。 1创建目标路径 由于DataX同步任务要求目标路径提前存在故需手动创建路径当前user表的目标路径应为/origin_data/medical/user_full/2023-05-09。
[atguiguhadoop102 bin]$ hadoop fs -mkdir -p /origin_data/medical/user_full/2023-05-092执行DataX同步命令
[atguiguhadoop102 bin]$ python /opt/module/datax/bin/datax.py -p-Dtargetdir/origin_data/medical/user_full/2023-05-09 /opt/module/datax/job/medical/import/medical.user.json3观察同步结果 观察HFDS目标路径是否出现数据。
全量表数据同步脚本
为方便使用以及后续的任务调度此处编写一个全量表数据同步脚本。 1在~/bin目录创建medical_mysql_to_hdfs_full.sh [atguiguhadoop102 bin]$ vim ~/bin/medical_mysql_to_hdfs_full.sh
#!/bin/bashDATAX_HOME/opt/module/datax
DATAX_DATA/opt/module/datax/job/medical#清理脏数据
handle_targetdir() {hadoop fs -rm -r $1 /dev/null 21hadoop fs -mkdir -p $1
}#数据同步
import_data() {local datax_config$1local target_dir$2handle_targetdir $target_direcho 正在处理$1python $DATAX_HOME/bin/datax.py -p-Dtargetdir$target_dir $datax_config /tmp/datax_run.log 21if [ $? -ne 0 ]thenecho 处理失败, 日志如下:cat /tmp/datax_run.log fi
}#接收表名变量
tab$1
# 如果传入日期则do_date等于传入的日期否则等于前一天日期
if [ -n $2 ] ;thendo_date$2
elsedo_date$(date -d -1 day %F)
ficase ${tab} in
dict|doctor|hospital|medicine|patient|user)import_data $DATAX_DATA/import/medical.${tab}.json /origin_data/medical/${tab}_full/$do_date;;
all)for tmp in dict doctor hospital medicine patient userdoimport_data $DATAX_DATA/import/medical.${tmp}.json /origin_data/medical/${tmp}_full/$do_datedone;;
esac2为mysql_to_hdfs_full.sh增加执行权限 [atguiguhadoop102 bin]$ chmod x ~/bin/medical_mysql_to_hdfs_full.sh 3测试同步脚本 [atguiguhadoop102 bin]$ medical_mysql_to_hdfs_full.sh all 2023-05-09 4检查同步结果 查看HDFS目表路径是否出现全量表数据全量表共6张。 全量表同步逻辑比较简单只需每日执行全量表数据同步脚本medical_mysql_to_hdfs_full.sh即可。
增量表数据同步 MySQL - Maxwell - Kafka - Flume - HDFS
数据通道 Flume配置 1Flume配置概述 Flume需要将Kafka中各topic的数据传输到HDFS因此选用KafkaSource以及HDFSSink。对于安全性要求高的数据不允许丢失选用FileChannel允许部分丢失的数据如日志可以选用MemoryChannel以追求更高的效率。此处采集的是业务数据不允许丢失选用FileChannel生产环境根据实际情况选择合适的组件。
KafkaSource订阅Kafka medical_ods主题的数据HDFSSink将不同topic的数据写入不同路径路径中应包含表名及日期前者用于区分来源于不同业务表的数据后者按天对数据进行划分。关键配置如下 具体数据示例如下
链接: 医疗数仓配置Flume
Maxwell配置
1Maxwell时间戳问题 为了让Maxwell时间戳日期与模拟的业务日期保持一致对Maxwell源码进行改动增加了mock_date参数在/opt/module/maxwell/config.properties文件中将该参数的值修改为业务日期即可。
2补充mock.date参数 配置参数如下。
log_levelinfo#Maxwell数据发送目的地可选配置有stdout|file|kafka|kinesis|pubsub|sqs|rabbitmq|redis
producerkafka
# 目标Kafka集群地址
kafka.bootstrap.servershadoop102:9092,hadoop103:9092,hadoop104:9092
#目标Kafka topic可静态配置例如:maxwell也可动态配置例如%{database}_%{table}
kafka_topictopic_db# MySQL相关配置
hosthadoop102
usermaxwell
passwordmaxwell
jdbc_optionsuseSSLfalseserverTimezoneAsia/ShanghaiallowPublicKeyRetrievaltrue# 指定数据按照主键分组进入Kafka不同分区避免数据倾斜
producer_partition_byprimary_key# 修改数据时间戳的日期部分
mock_date2023-05-09
3重新启动Maxwell [atguiguhadoop102 bin]$ mxw.sh restart 4重新生成模拟数据 [atguiguhadoop102 bin]$ medical_mock.sh 1 5观察HDFS目标路径日期是否与业务日期保持一致
增量表首日全量同步
通常情况下增量表需要在首日进行一次全量同步后续每日再进行增量同步首日全量同步可以使用Maxwell的bootstrap功能方便起见下面编写一个增量表首日全量同步脚本。 1在~/bin目录创建medical_mysql_to_kafka_inc_init.sh [atguiguhadoop102 bin]$ vim medical_mysql_to_kafka_inc_init.sh 脚本内容如下
#!/bin/bash# 该脚本的作用是初始化所有的增量表只需执行一次MAXWELL_HOME/opt/module/maxwellimport_data() {for tab in $do $MAXWELL_HOME/bin/maxwell-bootstrap --database medical --table $tab --config $MAXWELL_HOME/config.propertiesdone
}case $1 in
consultation | payment | prescription | prescription_detail | user | patient | doctor)import_data $1;;
all)import_data consultation payment prescription prescription_detail user patient doctor;;
esac2为medical_mysql_to_kafka_inc_init.sh增加执行权限 [atguiguhadoop102 bin]$ chmod 777 ~/bin/medical_mysql_to_kafka_inc_init.sh 3测试同步脚本 1清理历史数据 为方便查看结果现将HDFS上之前同步的增量表数据删除。 [atguiguhadoop102 ~]$ hadoop fs -ls /origin_data/medical | grep _inc | awk {print KaTeX parse error: Expected EOF, got } at position 2: 8}̲ | xargs hadoo… medical_mysql_to_kafka_inc_init.sh all 4检查同步结果 观察HDFS上是否重新出现增量表数据。
