离线数仓同步数据3

业务数据_增量表数据同步

  • 1)Flume配置概述
  • 2)Flume配置实操
  • 3)通道测试
  • 4)编写Flume启停脚本

在这里插入图片描述

1)Flume配置概述

Flume需要将Kafka中topic_db主题的数据传输到HDFS,故其需选用KafkaSource以及HDFSSink,Channel选用FileChannel。
需要注意的是, HDFSSink需要将不同mysql业务表的数据写到不同的路径,并且路径中应当包含一层日期,用于区分每天的数据。关键配置如下:

2)Flume配置实操

1)创建Flume配置文件
在hadoop104节点的Flume的job目录下创建kafka_to_hdfs_db.conf
[atguigu@hadoop104 flume]$ mkdir job
[atguigu@hadoop104 flume]$ vim job/kafka_to_hdfs_db.conf 
(2)配置文件内容如下
a1.sources = r1
a1.channels = c1
a1.sinks = k1a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.r1.batchSize = 5000
a1.sources.r1.batchDurationMillis = 2000
a1.sources.r1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092
a1.sources.r1.kafka.topics = topic_db
a1.sources.r1.kafka.consumer.group.id = flume
a1.sources.r1.setTopicHeader = true
a1.sources.r1.topicHeader = topic
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.atguigu.gmall.flume.interceptor.TimestampAndTableNameInterceptor$Buildera1.channels.c1.type = file
a1.channels.c1.checkpointDir = /opt/module/flume/checkpoint/behavior2
a1.channels.c1.dataDirs = /opt/module/flume/data/behavior2/
a1.channels.c1.maxFileSize = 2146435071
a1.channels.c1.capacity = 1000000
a1.channels.c1.keep-alive = 6## sink1
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /origin_data/gmall/db/%{tableName}_inc/%Y-%m-%d
a1.sinks.k1.hdfs.filePrefix = db
a1.sinks.k1.hdfs.round = falsea1.sinks.k1.hdfs.rollInterval = 10
a1.sinks.k1.hdfs.rollSize = 134217728
a1.sinks.k1.hdfs.rollCount = 0a1.sinks.k1.hdfs.fileType = CompressedStream
a1.sinks.k1.hdfs.codeC = gzip## 拼装
a1.sources.r1.channels = c1
a1.sinks.k1.channel= c1(3)编写Flume拦截器
新建一个Maven项目,并在pom.xml文件中加入如下配置
<dependencies><dependency><groupId>org.apache.flume</groupId><artifactId>flume-ng-core</artifactId><version>1.9.0</version><scope>provided</scope></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.62</version></dependency>
</dependencies><build><plugins><plugin><artifactId>maven-compiler-plugin</artifactId><version>2.3.2</version><configuration><source>1.8</source><target>1.8</target></configuration></plugin><plugin><artifactId>maven-assembly-plugin</artifactId><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs></configuration><executions><execution><id>make-assembly</id><phase>package</phase><goals><goal>single</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.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;
public class TimestampAndTableNameInterceptor implements Interceptor {@Overridepublic void initialize() {}@Overridepublic Event intercept(Event event) {Map<String, String> headers = event.getHeaders();
String log = new String(event.getBody(), StandardCharsets.UTF_8);JSONObject jsonObject = JSONObject.parseObject(log);Long ts = jsonObject.getLong("ts");//Maxwell输出的数据中的ts字段时间戳单位为秒,Flume HDFSSink要求单位为毫秒String timeMills = String.valueOf(ts * 1000);String tableName = jsonObject.getString("table");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) {}}
}重新打包
将打好的包放入到hadoop104的/opt/module/flume/lib文件夹下
[atguigu@hadoop102 lib]$ ls | grep interceptor
flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar

3)通道测试

1)启动Zookeeper、Kafka集群
(2)启动hadoop104的Flume
[atguigu@hadoop104 flume]$ bin/flume-ng agent -n a1 -c conf/ -f job/kafka_to_hdfs_db.conf -Dflume.root.logger=info,console
(3)生成模拟数据
[atguigu@hadoop102 bin]$ cd /opt/module/db_log/
[atguigu@hadoop102 db_log]$ java -jar gmall2020-mock-db-2021-11-14.jar 
(4)观察HDFS上的目标路径是否有数据出现
若HDFS上的目标路径已有增量表的数据出现了,就证明数据通道已经打通。
(5)数据目标路径的日期说明
仔细观察,会发现目标路径中的日期,并非模拟数据的业务日期,而是当前日期。这是由于Maxwell输出的JSON字符串中的ts字段的值,是数据的变动日期。而真实场景下,数据的业务日期与变动日期应当是一致的。

4)编写Flume启停脚本

为方便使用,此处编写一个Flume的启停脚本
(1)在hadoop102节点的/home/atguigu/bin目录下创建脚本f3.sh
[atguigu@hadoop102 bin]$ vim f3.sh在脚本中填写如下内容
#!/bin/bashcase $1 in
"start")echo " --------启动 hadoop104 业务数据flume-------"ssh hadoop104 "nohup /opt/module/flume/bin/flume-ng agent -n a1 -c /opt/module/flume/conf -f /opt/module/flume/job/kafka_to_hdfs_db.conf >/dev/null 2>&1 &"
;;
"stop")echo " --------停止 hadoop104 业务数据flume-------"ssh hadoop104 "ps -ef | grep kafka_to_hdfs_db | grep -v grep |awk '{print \$2}' | xargs -n1 kill"
;;
esac
(2)增加脚本执行权限
[atguigu@hadoop102 bin]$ chmod 777 f3.sh
(3)f3启动
[atguigu@hadoop102 module]$ f3.sh start
(4)f3停止
[atguigu@hadoop102 module]$ f3.sh stop
2.2.6.3 Maxwell配置
1)Maxwell时间戳问题此处为了模拟真实环境,对Maxwell源码进行了改动,增加了一个参数mock_date,该参数的作用就是指定Maxwell输出JSON字符串的ts时间戳的日期,接下来进行测试。
修改Maxwell配置文件config.properties,增加mock_date参数,如下
log_level=infoproducer=kafka
kafka.bootstrap.servers=hadoop102:9092,hadoop103:9092#kafka topic配置
kafka_topic=topic_db#注:该参数仅在maxwell教学版中存在,修改该参数后重启Maxwell才可生效
mock_date=2020-06-14# mysql login info
host=hadoop102
user=maxwell
password=maxwell
jdbc_options=useSSL=false&serverTimezone=Asia/Shanghai
注:该参数仅供学习使用,修改该参数后重启Maxwell才可生效。
重启Maxwell
[atguigu@hadoop102 bin]$ mxw.sh restart
重新生成模拟数据
[atguigu@hadoop102 bin]$ cd /opt/module/db_log/
[atguigu@hadoop102 db_log]$ java -jar gmall2020-mock-db-2021-11-14.jar 
观察HDFS目标路径日期是否正常

2.2.6.4 增量表首日全量同步

通常情况下,增量表需要在首日进行一次全量同步,后续每日再进行增量同步,首日全量同步可以使用Maxwell的bootstrap功能,方便起见,下面编写一个增量表首日全量同步脚本。
1)在~/bin目录创建mysql_to_kafka_inc_init.sh
[atguigu@hadoop102 bin]$ vim mysql_to_kafka_inc_init.sh
脚本内容如下
#!/bin/bash# 该脚本的作用是初始化所有的增量表,只需执行一次MAXWELL_HOME=/opt/module/maxwellimport_data() {$MAXWELL_HOME/bin/maxwell-bootstrap --database gmall --table $1 --config $MAXWELL_HOME/config.properties
}case $1 in
"cart_info")import_data cart_info;;
"comment_info")import_data comment_info;;
"coupon_use")import_data coupon_use;;
"favor_info")import_data favor_info;;
"order_detail")import_data order_detail;;
"order_detail_activity")import_data order_detail_activity;;
"order_detail_coupon")import_data order_detail_coupon;;
"order_info")import_data order_info;;
"order_refund_info")import_data order_refund_info;;
"order_status_log")import_data order_status_log;;
"payment_info")import_data payment_info;;
"refund_payment")import_data refund_payment;;
"user_info")import_data user_info;;
"all")import_data cart_infoimport_data comment_infoimport_data coupon_useimport_data favor_infoimport_data order_detailimport_data order_detail_activityimport_data order_detail_couponimport_data order_infoimport_data order_refund_infoimport_data order_status_logimport_data payment_infoimport_data refund_paymentimport_data user_info;;
esac
2)为mysql_to_kafka_inc_init.sh增加执行权限
[atguigu@hadoop102 bin]$ chmod 777 ~/bin/mysql_to_kafka_inc_init.sh

3)测试同步脚本
(1)清理历史数据
为方便查看结果,现将HDFS上之前同步的增量表数据删除
[atguigu@hadoop102 ~]$ hadoop fs -ls /origin_data/gmall/db | grep _inc | awk '{print KaTeX parse error: Expected 'EOF', got '}' at position 2: 8}̲' | xargs hadoo… mysql_to_kafka_inc_init.sh all
4)检查同步结果
观察HDFS上是否重新出现增量表数据。


2.3 采集通道启动/停止脚本
1)在/home/atguigu/bin目录下创建脚本cluster.sh
[atguigu@hadoop102 bin]$ vim cluster.sh在脚本中填写如下内容
#!/bin/bashcase $1 in
"start"){echo ================== 启动 集群 ==================#启动 Zookeeper集群zk.sh start#启动 Hadoop集群hdp.sh start#启动 Kafka采集集群kf.sh start#启动采集 Flumef1.sh start#启动日志消费 Flumef2.sh start#启动业务消费 Flumef3.sh start#启动 maxwellmxw.sh start};;
"stop"){echo ================== 停止 集群 ==================#停止 Maxwellmxw.sh stop#停止 业务消费Flumef3.sh stop#停止 日志消费Flumef2.sh stop#停止 日志采集Flumef1.sh stop#停止 Kafka采集集群kf.sh stop#停止 Hadoop集群hdp.sh stop#停止 Zookeeper集群zk.sh stop};;
esac2)增加脚本执行权限
[atguigu@hadoop102 bin]$ chmod 777 cluster.sh
3)cluster集群启动脚本
[atguigu@hadoop102 module]$ cluster.sh start
4)cluster集群停止脚本
[atguigu@hadoop102 module]$ cluster.sh stop

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/72251.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

内网穿透实现Windows远程桌面访问Ubuntu,简单高效的远程桌面解决方案

文章目录 前言1. ubuntu安装XRDP2.局域网测试连接3.安装cpolar内网穿透4.cpolar公网地址测试访问5.固定域名公网地址 前言 XRDP是一种开源工具&#xff0c;它允许用户通过Windows RDP访问Linux远程桌面。 除了Windows RDP外&#xff0c;xrdp工具还接受来自其他RDP客户端(如Fre…

uni-app 之 uni.request 网络请求API接口

uni-app 之 uni.request 网络请求API接口 image.png <template><!-- vue2的<template>里必须要有一个盒子&#xff0c;不能有两个&#xff0c;这里的盒子就是 view--><view>--- uni.request 网络请求API接口 ---<view><!-- 免费的测试接口 --…

制造企业如何优化物料控制?

导 读 ( 文/ 2127 ) 物料控制是指对制造过程中所涉及的物料流动和库存进行有效管理和控制的过程。它包括物料需求计划、供应商管理、物料采购、物料接收和入库、物料库存管理以及物料发放和使用等关键环节。通过精确的物料需求计划和库存管理&#xff0c;物料控制可以确保物料供…

转载: 又拍云【PrismCDN 】低延时的P2P HLS直播技术实践

低延时的P2P HLS直播技术实践本文是第二部分《PrismCDN 网络的架构解析,以及低延迟、低成本的奥秘》低延时的P2P HLS直播技术实践 [首页 > Open Talk NO.41 | 2018 音视频技术沙龙深圳站 > 低延时 WebP2P 直播技术实践https://opentalk-blog.b0.upaiyun.com/prod/2018-0…

【群答疑】jmeter关联获取上一个请求返回的字符串,分割后保存到数组,把数组元素依次作为下一个请求的入参...

一个非常不错的问题&#xff0c;来检验下自己jmeter基本功 可能有同学没看懂题&#xff0c;这里再解释一下&#xff0c;上面问题需求是&#xff1a;jmeter关联获取上一个请求返回的字符串&#xff0c;分割后保存到数组&#xff0c;把数组元素依次作为下一个请求的入参 建议先自…

TCP流量控制和拥塞控制,具体在场景中是怎么起作用的

TCP的流量控制 所谓的流量控制就是让发送方的发送速率不要太快&#xff0c;让接收方来得及接受。利用滑动窗口机制可以很方便的在TCP连接上实现对发送方的流量控制。TCP的窗口单位是字节&#xff0c;不是报文段&#xff0c;发送方的发送窗口不能超过接收方给出的接收窗口的数值…

《TCP/IP网络编程》阅读笔记--域名及网络地址

目录 1--域名系统 2--域名与 IP 地址的转换 2-1--利用域名来获取 IP 地址 2-2--利用 IP 地址获取域名 3--代码实例 3-1--gethostbyname() 3-2--gethostbyaddr() 1--域名系统 域名系统&#xff08;Domain Name System&#xff0c;DNS&#xff09;是对 IP 地址和域名进行相…

OLED透明屏模块:引领未来显示技术的突破

OLED透明屏模块作为一项引领未来显示技术的突破&#xff0c;以其独特的特点和卓越的画质在市场上引起了广泛关注。 根据行业报告&#xff0c;预计到2025年&#xff0c;OLED透明屏模块将占据智能手机市场的20%份额&#xff0c;并在汽车导航系统市场中占据30%以上份额。 那么&am…

安装K8s基础环境软件(二)

所有节点执行 1、安装docker sudo yum install -y yum-utils sudo yum-config-manager --add-repo https://download.docker.com/linux/centos/docker-ce.reposudo yum install docker-ce docker-ce-cli containerd.io docker-buildx-plugin docker-compose-plugin systemctl…

Python web 框架web.py「简约美」

web.py is a web framework for Python that is as simple as it is powerful. web.py is in the public domain, you can use it for whatever purpose with absolutely no restrictions. web.py 是一个简单而强大的 Python Web 框架。web.py 属于公共领域&#xff0c;您可以…

《C++设计模式》——行为型

前言 行为型模式是对在不同的对象之间划分责任和算法的抽象化。行为型模式不仅仅关注类和对象的结构&#xff0c;而且重点关注它们之间的相互作用。 Interpreter(解释器) Template Method(模板方法) GOOD&#xff1a;把不变的代码部分都转移到父类中&#xff0c;将可变的代…

【深度学习实验】数据可视化

目录 一、实验介绍 二、实验环境 三、实验内容 0. 导入库 1. 归一化处理 归一化 实验内容 2. 绘制归一化数据折线图 报错 解决 3. 计算移动平均值SMA 移动平均值 实验内容 4. 绘制移动平均值折线图 5 .同时绘制两图 6. array转换为tensor张量 7. 打印张量 一、…

数据结构与算法:练习与实践的重要性

文章目录 为什么练习与实践很重要&#xff1f;1. 熟练应用2. 问题解决能力3. 代码效率4. 面试准备 如何练习与实践&#xff1f;1. 在线评测平台2. 自主设计数据结构3. 解决不同类型的问题 持续学习与实践 &#x1f389;欢迎来到数据结构学习专栏~数据结构与算法&#xff1a;练习…

【C++进阶(五)】STL大法--list模拟实现以及list和vector的对比

&#x1f493;博主CSDN主页:杭电码农-NEO&#x1f493;   ⏩专栏分类:C从入门到精通⏪   &#x1f69a;代码仓库:NEO的学习日记&#x1f69a;   &#x1f339;关注我&#x1faf5;带你学习C   &#x1f51d;&#x1f51d; list模拟实现 1. 前言2. list类的大致框架与结构…

Linux下的系统编程——共享存储映射(十)

前言&#xff1a; mmap是一种内存映射文件的方法&#xff0c;即将一个文件或者其它对象映射到进程的地址空间&#xff0c;实现文件磁盘地址和进程虚拟地址空间中一段虚拟地址的一一对映关系。实现这样的映射关系后&#xff0c;进程就可以采用指针的方式读写操作这一段内存&…

车船边缘网关是如何给车辆船只定位的?

随着智能交通系统的不断发展&#xff0c;车路协同成为了重要的研究方向之一。而AI边缘计算网关在这个领域中发挥着至关重要的作用。本文将重点介绍AI边缘计算网关在车路协同中的应用&#xff0c;并强调其中的重点词汇或短语。 首先&#xff0c;什么是AI边缘计算网关&#xff1…

使用命令行创建仓库

如果你还没有任何代码&#xff0c;可以通过命令行工具创建一个全新的Git仓库并初始化到本项目仓库中。 git clone https://e.coding.net/***/neurosens.git cd neurosens echo "# neurosens" >> README.md git add README.md git commit -m "first commi…

2023年09月编程语言流行度排名

点击查看最新编程语言流行度排名&#xff08;每月更新&#xff09; 2023年09月编程语言流行度排名 编程语言流行度排名是通过分析在谷歌上搜索语言教程的频率而创建的 一门语言教程被搜索的次数越多&#xff0c;大家就会认为该语言越受欢迎。这是一个领先指标。原始数据来自…

解决windows下git操作提示用户名密码错误的问题

当代码从一个平台切换到另一个平台的时候&#xff0c;需要做两步操作&#xff0c;第一步就是更新git的仓库地址&#xff0c;在项目的.git/config文件里面修改&#xff0c;这一步做完之后&#xff0c;就可以推送代码到新的仓库了&#xff0c;这里就是重点来了。 一般第一次推动代…

27.方向标

题目 描述 一位木匠收到了一个木制指示牌的订单。每块木板必须与前一块垂直对齐&#xff0c;要么与前一个箭头的基部对齐&#xff0c;要么与相反的一侧对齐&#xff0c;在那里用特制的螺钉固定。两块木板必须重叠。木匠将设计师发送的草图编码成了一个整数序列&#xff0c;但…