文章目录
- 一、全量同步
- 1.1 梳理需要全量同步的业务表
- 1.2 Sqoop: MySQL To HDFS
- 1.2.1 开发脚本
- 1.2.2 授予脚本执行权限
- 1.2.3 执行脚本
- 1.2.4 定时调度
- 二、增量同步
- 2.1 梳理需要增量同步的业务表
- 2.2 Maxwell: MySQL To Kafka
- 2.2.1 首次全量同步
- 2.2.2 每日增量同步
- 2.2.2.1 编写Maxwell配置文件
- 2.2.2.2 启动Maxwell
- 2.3 Flume: Kafka To HDFS
- 2.3.1 自定义拦截器
- 2.3.2 编写配置文件
- 2.3.3 创建Kafka Topic
- 2.3.4 启动/停止Flume
- 三、正式上线
- 3.1 删除HDFS上业务测试数据
- 3.2 删除Hive外部表对应HDFS的数据
- 3.3 Sqoop全量同步
- 3.3.1 首次执行
- 3.3.2 定时调度
- 3.4 Maxwell全量同步
一、全量同步
数据流转：MySQL ---------Sqoop脚本---------》 HDFS
1.1 梳理需要全量同步的业务表
| 序号 | 表名 | 表中文名 |
|---|---|---|
| 1 | sku_info | 库存单元表 |
| 2 | base_category1 | 一级分类表 |
| 3 | base_category2 | 二级分类表 |
| 4 | base_category3 | 三级分类表 |
| 5 | base_province | 省份表 |
| 6 | base_trademark | 品牌表 |
| 7 | spu_info | 商品表 |
| 8 | favor_info | 商品收藏表 |
| 9 | cart_info | 购物车表 |
| 10 | coupon_info | 优惠券表 |
| 11 | activity_info | 活动表 |
| 12 | activity_rule | 优惠规则 |
| 13 | base_dic | 字典表 |
| 14 | sku_attr_value | sku平台属性值关联表 |
| 15 | sku_sale_attr_value | sku销售属性值 |
| 16 | base_region | 地区表 |
| 17 | user_info | 用户表 |
| 18 | order_info | 订单表 |
| 19 | coupon_use | 优惠券领用表 |
| 20 | order_status_log | 订单状态日志表 |
| 21 | order_detail | 订单明细表 |
| 22 | payment_info | 支付信息表 |
| 23 | comment_info | 商品评论表 |
| 24 | order_refund_info | 退单表 |
| 25 | order_detail_activity | 订单明细活动表 |
| 26 | order_detail_coupon | 订单明细购物券表 |
| 27 | refund_payment | 退款信息表 |
1.2 Sqoop: MySQL To HDFS
1.2.1 开发脚本
脚本放在/home/hadoop/bin目录下的原因：该脚本是给普通用户hadoop使用的，在普通用户hadoop登陆的情况下执行echo $PATH命令，发现/home/hadoop/bin在其中，表示普通用户hadoop可以任意地方使用该目录下的可执行文件。
[hadoop@hadoop102 ~]$ cd /home/hadoop/
[hadoop@hadoop102 ~]$ mkdir bin
[hadoop@hadoop102 ~]$ vim /home/hadoop/bin/mysql_to_hdfs_full.sh
脚本内容：
–delete-target-dir：如果目标文件夹存在，先删除后插入。
#! /bin/bashif [ -n "$2" ] ;thendo_date=$2
elsedo_date=`date -d '-1 day' +%F`
fiimport_data(){
/opt/module/sqoop-1.4.6.bin__hadoop-2.0.4-alpha/bin/sqoop import \
--connect jdbc:mysql://mall:3306/gmall \
--username root \
--password 123456 \
--target-dir /warehouse/db/gmall/$1_full/$do_date \
--delete-target-dir \
--query "$2 and \$CONDITIONS" \
--num-mappers 1 \
--fields-terminated-by '\t' \
--compress \
--compression-codec gzip \
--null-string '\\N' \
--null-non-string '\\N'
}import_sku_info(){import_data "sku_info" "select id,spu_id,price,sku_name,sku_desc,weight,tm_id,category3_id,is_sale,create_timefrom sku_info where 1=1"
}import_base_category1(){import_data "base_category1" "select id,name from base_category1 where 1=1"
}import_base_category2(){import_data "base_category2" "selectid,name,category1_id from base_category2 where 1=1"
}import_base_category3(){import_data "base_category3" "selectid,name,category2_idfrom base_category3 where 1=1"
}import_base_province(){import_data "base_province" "selectid,name,region_id,area_code,iso_code,iso_3166_2from base_provincewhere 1=1"
}import_base_region(){import_data "base_region" "selectid,region_namefrom base_regionwhere 1=1"
}import_base_trademark(){import_data "base_trademark" "selectid,tm_namefrom base_trademarkwhere 1=1"
}import_spu_info(){import_data "spu_info" "selectid,spu_name,category3_id,tm_idfrom spu_infowhere 1=1"
}import_favor_info(){import_data "favor_info" "selectid,user_id,sku_id,spu_id,is_cancel,create_time,cancel_timefrom favor_infowhere 1=1"
}import_cart_info(){import_data "cart_info" "selectid,user_id,sku_id,cart_price,sku_num,sku_name,create_time,operate_time,is_ordered,order_time,source_type,source_idfrom cart_infowhere 1=1"
}import_coupon_info(){import_data "coupon_info" "selectid,coupon_name,coupon_type,condition_amount,condition_num,activity_id,benefit_amount,benefit_discount,create_time,range_type,limit_num,taken_count,start_time,end_time,operate_time,expire_timefrom coupon_infowhere 1=1"
}import_activity_info(){import_data "activity_info" "selectid,activity_name,activity_type,start_time,end_time,create_time,activity_descfrom activity_infowhere 1=1"
}import_activity_rule(){import_data "activity_rule" "selectid,activity_id,activity_type,condition_amount,condition_num,benefit_amount,benefit_discount,benefit_levelfrom activity_rulewhere 1=1"
}import_base_dic(){import_data "base_dic" "selectdic_code,dic_name,parent_code,create_time,operate_timefrom base_dicwhere 1=1"
}import_sku_attr_value(){import_data "sku_attr_value" "selectid,attr_id,value_id,sku_id,attr_name,value_namefrom sku_attr_valuewhere 1=1"
}import_sku_sale_attr_value(){import_data "sku_sale_attr_value" "selectid,sku_id,spu_id,sale_attr_value_id,sale_attr_id,sale_attr_name,sale_attr_value_namefrom sku_sale_attr_valuewhere 1=1"
}import_user_info(){import_data "user_info" "selectid,login_name,nick_name,passwd,name,phone_num,email,head_img,user_level,birthday,gender,create_time,operate_time,statusfrom user_infowhere 1=1"
}import_order_info(){import_data "order_info" "selectid,consignee,consignee_tel,total_amount,order_status,user_id,payment_way,delivery_address,order_comment,out_trade_no,trade_body,create_time,operate_time,expire_time,process_status,tracking_no,parent_order_id,img_url,province_id,activity_reduce_amount,coupon_reduce_amount,original_total_amount,feight_fee,feight_fee_reduce,refundable_timefrom order_infowhere 1=1"
}import_coupon_use(){import_data "coupon_use" "selectid,coupon_id,user_id,order_id,coupon_status,get_time,using_time,used_time,expire_timefrom coupon_usewhere 1=1"
}import_order_status_log(){import_data "order_status_log" "selectid,order_id,order_status,operate_timefrom order_status_logwhere 1=1"
}import_order_detail(){import_data "order_detail" "selectid,order_id,sku_id,sku_name,img_url,order_price,sku_num,create_time,source_type,source_id,split_total_amount,split_activity_amount,split_coupon_amountfrom order_detailwhere 1=1"
}import_payment_info(){import_data "payment_info" "selectid,out_trade_no,order_id,user_id,payment_type,trade_no,total_amount,subject,payment_status,create_time,callback_time,callback_contentfrom payment_infowhere 1=1"
}import_comment_info(){import_data "comment_info" "selectid,user_id,nick_name,head_img,sku_id,spu_id,order_id,appraise,comment_txt,create_time,operate_timefrom comment_infowhere 1=1"
}import_order_refund_info(){import_data "order_refund_info" "selectid,user_id,order_id,sku_id,refund_type,refund_num,refund_amount,refund_reason_type,refund_reason_txt,refund_status,create_timefrom order_refund_infowhere 1=1"
}import_order_detail_activity(){import_data "order_detail_activity" "selectid,order_id,order_detail_id,activity_id,activity_rule_id,sku_id,create_timefrom order_detail_activitywhere 1=1"
}import_order_detail_coupon(){import_data "order_detail_coupon" "selectid,order_id,order_detail_id,coupon_id,coupon_use_id,sku_id,create_timefrom order_detail_couponwhere 1=1"
}import_refund_payment(){import_data "refund_payment" "selectid,out_trade_no,order_id,sku_id,payment_type,trade_no,total_amount,subject,refund_status,create_time,callback_time,callback_contentfrom refund_paymentwhere 1=1"
}case $1 in"sku_info")import_sku_info
;;"base_category1")import_base_category1
;;"base_category2")import_base_category2
;;"base_category3")import_base_category3
;;"base_province")import_base_province
;;"base_region")import_base_region
;;"base_trademark")import_base_trademark
;;"spu_info")import_spu_info
;;"favor_info")import_favor_info
;;"cart_info")import_cart_info
;;"coupon_info")import_coupon_info
;;"activity_info")import_activity_info
;;"activity_rule")import_activity_rule
;;"base_dic")import_base_dic
;;"sku_attr_value")import_sku_attr_value
;;"sku_sale_attr_value")import_sku_sale_attr_value
;;"user_info")import_user_info
;;"order_info")import_order_info
;;"coupon_use")import_coupon_use
;;"order_status_log")import_order_status_log
;;"order_detail")import_order_detail
;;"payment_info")import_payment_info
;;"comment_info")import_comment_info
;;"order_refund_info")import_order_refund_info
;;"order_detail_activity")import_order_detail_activity
;;"order_detail_coupon")import_order_detail_coupon
;;"refund_payment")import_refund_payment
;;
"all")import_sku_infoimport_base_category1import_base_category2import_base_category3import_base_provinceimport_base_regionimport_base_trademarkimport_spu_infoimport_favor_infoimport_cart_infoimport_coupon_infoimport_activity_infoimport_activity_ruleimport_base_dicimport_sku_attr_valueimport_sku_sale_attr_valueimport_user_infoimport_order_infoimport_coupon_useimport_order_status_logimport_order_detailimport_payment_infoimport_comment_infoimport_order_refund_infoimport_order_detail_activityimport_order_detail_couponimport_refund_payment
;;
esac
1.2.2 授予脚本执行权限
[hadoop@hadoop102 bin]$ chmod +x mysql_to_hdfs_full.sh
1.2.3 执行脚本
每天凌晨执行不应该带上日期参数。
[hadoop@hadoop102 ~]$ mysql_to_hdfs_full.sh all 2023-12-02
1.2.4 定时调度
凌晨执行，拉取MySQL全量数据放入HDFS昨天分区。
二、增量同步
数据流转：MySQL ---------Maxwell监控程序--------》 Kafka ----------Flume监控程序---------》 HDFS
怎么处理删除和更新？？？？？？
2.1 梳理需要增量同步的业务表
| 序号 | 表名 | 表中文名 |
|---|---|---|
| 1 | order_info | 订单表 |
| 2 | coupon_use | 优惠券领用表 |
| 3 | order_status_log | 订单状态日志表 |
| 4 | user_info | 用户信息表 |
| 5 | order_detail | 订单明细表 |
| 6 | payment_info | 支付信息表 |
| 7 | comment_info | 商品评论表 |
| 8 | order_refund_info | 退单表 |
| 9 | order_detail_activity | 订单明细活动表 |
| 10 | order_detail_coupon | 订单明细购物券表 |
| 11 | refund_payment | 退款信息表 |
2.2 Maxwell: MySQL To Kafka
2.2.1 首次全量同步
首次全量同步：使用maxwell-bootstrap功能将MySQL历史全量数据导入Kafka。
注意：增量表全量初始化前不要开启对该表的增量Maxwell采集，否则会造成该表在Kafka中的数据重复！初始化完成后再开启对该表的增量监控！
[mall@mall ~]$ mkdir bin
[mall@mall bin]$ vim /home/mall/bin/mysql_to_kafka_inc_init.sh
配置文件：
见下面2.2.2.1，与增量同步使用同一个配置文件。
脚本内容：
#!/bin/bash# 该脚本的作用是初始化增量表，只需执行一次import_data() {/opt/module/maxwell-1.29.2/bin/maxwell-bootstrap --database gmall --table $1 --config /opt/module/maxwell-1.29.2/config.properties
}case $1 in
"order_info")import_data order_info;;
"coupon_use")import_data coupon_use;;
"order_status_log")import_data order_status_log;;
"user_info")import_data user_info;;
"order_detail")import_data order_detail;;
"payment_info")import_data payment_info;;
"comment_info")import_data comment_info;;
"order_refund_info")import_data order_refund_info;;
"order_detail_activity")import_data order_detail_activity;;
"order_detail_coupon")import_data order_detail_coupon;;
"refund_payment")import_data refund_payment;;
"all")import_data order_infoimport_data coupon_useimport_data order_status_logimport_data user_infoimport_data order_detailimport_data payment_infoimport_data comment_infoimport_data order_refund_infoimport_data order_detail_activityimport_data order_detail_couponimport_data refund_payment;;
esac
授予脚本执行权限：
[mall@mall bin]$ chmod +x mysql_to_kafka_inc_init.sh
执行脚本：
[mall@mall ~]$ mysql_to_kafka_inc_init.sh all
2.2.2 每日增量同步
2.2.2.1 编写Maxwell配置文件
[mall@mall maxwell-1.29.2]$ vim config.properties
内容：
#Maxwell数据发送目的地，可选配置有stdout|file|kafka|kinesis|pubsub|sqs|rabbitmq|redis
producer=kafka
#目标Kafka集群地址
kafka.bootstrap.servers=hadoop102:9092,hadoop103:9092
#目标Kafka topic，可静态配置，例如:maxwell，也可动态配置，例如：%{database}_%{table}
kafka_topic=topic_mall_db_binlog#MySQL相关配置
host=mall
user=maxwell
password=maxwell
jdbc_options=useSSL=false&serverTimezone=Asia/Shanghai
2.2.2.2 启动Maxwell
# 启动
[mall@mall ~]$ /opt/module/maxwell-1.29.2/bin/maxwell --config /opt/module/maxwell-1.29.2/config.properties --daemon
# 停止
[mall@mall ~]$ ps -ef | grep maxwell | grep -v grep | awk '{print $2}'
[mall@mall ~]$ kill
2.3 Flume: Kafka To HDFS
2.3.1 自定义拦截器
套路：从body中拿出采集到的数据，解析出有用字段放入header中，配置文件中可以获取header中的东西。
作用1：把从Kafka中获取的json串的业务表名放到header中。
作用2：把从Kafka中获取的json串的ts时间戳转换成毫秒，放入header中。
代码：
package com.songshuang.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;/*** @date 2023/12/1 15:24*/
public class TimestampAndTableNameInterceptor implements Interceptor {@Overridepublic void initialize() {}@Overridepublic Event intercept(Event event) {// 1、从body中解析json拿出想要字段byte[] body = event.getBody();String db_binlog = new String(body, StandardCharsets.UTF_8);JSONObject jsonObject = JSONObject.parseObject(db_binlog);Long ts = jsonObject.getLong("ts");//Maxwell输出的数据中的ts字段时间戳单位为秒，Flume HDFSSink要求单位为毫秒String timeMills = String.valueOf(ts * 1000);String tableName = jsonObject.getString("table");// 2、放入header中Map<String, String> headers = event.getHeaders();headers.put("timestamp", timeMills);headers.put("tableName", tableName);return event;}@Overridepublic List<Event> intercept(List<Event> events) {for (Event event : events) {intercept(event);}return events;}@Overridepublic void close() {}// 建造者模式public static class Builder implements Interceptor.Builder {@Overridepublic Interceptor build() {return new TimestampAndTableNameInterceptor ();}@Overridepublic void configure(Context context) {}}
}
2.3.2 编写配置文件
[hadoop@hadoop104 ~]$ cd /opt/module/apache-flume-1.9.0-bin/job/
[hadoop@hadoop104 job]$ vim kafka_to_hdfs_db.conf
内容：
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1# 配置source
a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.r1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092
a1.sources.r1.kafka.topics = topic_mall_db_binlog
a1.sources.r1.kafka.consumer.group.id = consumer_group_flume_mall_db_binlog
# 指定consumer从哪个offset开始消费，默认latest
# a1.sources.r1.kafka.consumer.auto.offset.reset = earliest
# 自定义拦截器
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.songshuang.flume.interceptor.TimestampAndTableNameInterceptor$Builder# 配置sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /warehouse/db/gmall/%{tableName}_inc/%Y-%m-%d
a1.sinks.k1.hdfs.fileType = CompressedStream
a1.sinks.k1.hdfs.codeC = gzip
# 文件滚动策略
a1.sinks.k1.hdfs.rollInterval = 20
a1.sinks.k1.hdfs.rollSize = 134217728
a1.sinks.k1.hdfs.rollCount = 0# 配置channel
a1.channels.c1.type = file
a1.channels.c1.dataDirs = /opt/module/apache-flume-1.9.0-bin/data/kafka_to_hdfs_mall_db_binlog
a1.channels.c1.checkpointDir = /opt/module/apache-flume-1.9.0-bin/checkpoint/kafka_to_hdfs_mall_db_binlog# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
2.3.3 创建Kafka Topic
手动创建topic好处：可以自定义分区数、分区副本数。
[hadoop@hadoop102 ~]$ cd /opt/module/kafka_2.11-2.4.1/
[hadoop@hadoop102 kafka_2.11-2.4.1]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --create --replication-factor 2 --partitions 3 --topic topic_mall_db_binlog
2.3.4 启动/停止Flume
# 启动
[hadoop@hadoop104 job]$ cd /opt/module/apache-flume-1.9.0-bin/
[hadoop@hadoop104 apache-flume-1.9.0-bin]$ nohup bin/flume-ng agent --conf conf/ --name a1 --conf-file job/kafka_to_hdfs_db.conf &>/dev/null 2>/dev/null &
# 停止
[hadoop@hadoop104 apache-flume-1.9.0-bin]$ ps -ef | grep kafka_to_hdfs_db.conf | grep -v grep | awk '{print $2}'
[hadoop@hadoop104 apache-flume-1.9.0-bin]$ kill
三、正式上线
3.1 删除HDFS上业务测试数据
删除HDFS目录 /warehouse/db/gmall
3.2 删除Hive外部表对应HDFS的数据
删除HDFS目录 /warehouse/dw_ods.db/ods_业务表名
3.3 Sqoop全量同步
3.3.1 首次执行
注意：首次同步的数据不是很准确，分区数据可能错乱！
[hadoop@hadoop102 ~]$ mysql_to_hdfs_full.sh all
3.3.2 定时调度
默认给当前用户添加定时调度任务。
[hadoop@hadoop102 ~]$ crontab -e
内容：
每天的零点零分执行Sqoop全量同步脚本！
注意：添加要使用到的环境变量，最好与登陆用户的PATH保持一致！
原因：crontab用的自己的一套环境变量，并没有加载/etc/profile和~/.bash_profile等！
验证方式：分别通过命令行和crontab任务执行echo $PATH。
#JAVA_HOME
JAVA_HOME=/opt/module/jdk1.8.0_291
#HADOOP_HOME
HADOOP_HOME=/opt/module/hadoop-3.1.3
#KAFKA_HOME
KAFKA_HOME=/opt/module/kafka_2.11-2.4.1
# kafkaEagle
KE_HOME=/opt/module/kafka-eagle-web-1.4.8
#HIVE_HOME
HIVE_HOME=/opt/module/apache-hive-3.1.2-bin
#PATH
PATH=/usr/local/bin:/usr/bin:/usr/local/sbin:/usr/sbin:/opt/module/jdk1.8.0_291/bin:/opt/module/hadoop-3.1.3/bin:/opt/module/hadoop-3.1.3/sbin:/opt/module/kafka_2.11-2.4.1/bin:/opt/module/kafka-eagle-web-1.4.8/bin:/opt/module/apache-hive-3.1.2-bin/bin:/home/hadoop/.local/bin:/home/hadoop/bin0 0 * * * mysql_to_hdfs_full.sh all >/opt/module/sqoop-1.4.6.bin__hadoop-2.0.4-alpha/logs/crontab_mysql_to_hdfs_full_all.log 2>&1
查看当前用户具有的定时任务：
[hadoop@hadoop102 ~]$ crontab -l
查看crontab任务执行记录：
[hadoop@hadoop102 ~]$ sudo cat /var/log/cron | grep mysql_to_hdfs_full.sh
3.4 Maxwell全量同步
[mall@mall ~]$ mysql_to_kafka_inc_init.sh all
![[C++]内存管理：串顺序存储的内存回收](http://pic.xiahunao.cn/yaotu/[C++]内存管理：串顺序存储的内存回收)
