1. Toy模板网首页
  2. > Toy博客

【flume实时采集mysql数据库的数据到kafka】

9月前 作者：TKilome 分类：Toy博客 阅读(58) 违法举报

这篇具有很好参考价值的文章主要介绍了【flume实时采集mysql数据库的数据到kafka】。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

最近做了flume实时采集mysql数据到kafka的实验，做个笔记，防止忘记
！！！建议从头看到尾，因为一些简单的东西我在前面提了，后面没提。

Kafka搭建：https://blog.csdn.net/cjwfinal/article/details/120803013

flume搭建：https://blog.csdn.net/cjwfinal/article/details/120441503?spm=1001.2014.3001.5502

一、flume写入当前文件系统

题目：

编写配置文件，设置文件夹mylogs为source位置，文件夹backup为sink写入位置，实现对文件夹的数据备份。
新建两个文本文本文件1.txt与2.txt，在1.txt中输入Hello Flume，在2.txt中输入hello flume将两个文件拖入mylog，查看backup文件夹中出现的文件及其内容。文件可用记事本打开

解题：

1.flume配置文件

#a1表示agent的名字 可以自定义
# 给sources（在一个agent里可以定义多个source）取个名字
a1.sources = r1
# 给channel个名字
a1.channels = c1
# 给channel个名字
a1.sinks = k1

# 对source进行配置
# agent的名字.sources.source的名字.参数 = 参数值

# source的类型 spoolDir（监控一个目录下的文件的变化）
a1.sources.r1.type = spooldir
# 监听哪一个目录
a1.sources.r1.spoolDir = /root/mylogs
# 是否在event的headers中保存文件的绝对路径
a1.sources.r1.fileHeader = true
# 给拦截器取个名字 i1
a1.sources.r1.interceptors = i1
# 使用timestamp拦截器，将处理数据的时间保存到event的headers中
a1.sources.r1.interceptors.i1.type = timestamp

# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# 配置sink为logger
# 直接打印到控制台
a1.sinks.k1.type = file_roll
a1.sinks.k1.channel = c1
a1.sinks.k1.sink.directory = /root/backup

# 将source、channel、sink组装成agent
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

2.要保证配置文件中的文件的路径都存在，否则会报错

3.启动flume

启动命令：
flume-ng agent -n a1 -f spoolingtest.cong -Dflume.root.logger=DEBUG,console

a1是agent的名字，spoolingtest.cong是配置文件名，如果不一样请修改

【flume实时采集mysql数据库的数据到kafka】

4.测试

1.启动flume

注意：要在配置文件所在的目录启动

2.在mylogs文件夹下创建两个文件1.txt，2.txt，分别写入hello world和Hello World

写入并保存，后flume打印出来的日志会发生改变。

3.查看backup文件夹

backup文件夹中，会产生许多文件，其中某个文件中会保存上面两个文件的内容，自行查找即可。
【flume实时采集mysql数据库的数据到kafka】

二、flume连接kafka

题目

把flume监听到的文件内容，输入到，kafka，并由kafka的消费着消费出来。

解题

1.flume配置文件

 
# a1表示agent的名字 可以自定义
# # 给sources（在一个agent里可以定义多个source）取个名字
a1.sources = r1
# # 给channel个名字
a1.channels = c1
# # 给channel个名字
a1.sinks = k1
# 对source进行配置
# agent的名字.sources.source的名字.参数 = 参数值

# source的类型 spoolDir（监控一个目录下的文件的变化）
a1.sources.r1.type = exec
# 监听哪一个目录
a1.sources.r1.command = tail -F -c +0 /root/dir1/test.log
# 是否在event的headers中保存文件的绝对路径
a1.sources.r1.fileHeader = true
# 给拦截器取个名字 i1
a1.sources.r1.interceptors = i1
# 使用timestamp拦截器，将处理数据的时间保存到event的headers中
a1.sources.r1.interceptors.i1.type = timestamp

# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# 配置sinks，输出到kafka中
a1.sinks.k1.channel = c1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
# topic是指定kafka的主题
a1.sinks.k1.kafka.topic = events
# master是主机名，也可以是主机IP地址
a1.sinks.k1.kafka.bootstrap.servers = master:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1

# 将source、channel、sink组装成agent![在这里插入图片描述](https://img-blog.csdnimg.cn/13519938e4ba42b9a4294454f2013dc2.png#pic_center)

a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

2测试

1.创建指定的目录/root/dir1/test.log
2.启动flume
flume-ng agent -n a1 -f spoolingtest.cong -Dflume.root.logger=DEBUG,console
3.启动kafka的消费者
启动命令
#启动消费者之前首先要创建主题
kafka-topics.sh --create --topic events --bootstrap-server master:9092
#启动消费者
kafka-console-consumer.sh --bootstrap-server master:9092 --topic events --from-beginning
kafka主题操作命令
#删除主题
kafka-topics.sh --delete --topic events --zookeeper localhost:2181
#查看主题列表
kafka-topics.sh --list --bootstrap-server master:9092
4.在test.log中写入数据

【flume实时采集mysql数据库的数据到kafka】

5.观察kafka的消费是否消费出数据

【flume实时采集mysql数据库的数据到kafka】

三、flume实时采集mysql数据到kafka

题目

在MySQL中建立数据库school，在数据库中建立表student。SQL语句如下：

#创建数据库
create database school;
#使用数据库
use school;
#创建数据表
create table student(
	id int not null,
	name varchar(40),
	age int,
	grade int,
	primary key(id)
);
# 删除数据
drop database school;
# 删除数据表
drop database student;
# 查看数据表
show tables;

编写配置文件，将student表中的内容输出到控制台。启动Flume，在student表中使用下列命令插入数据，在Kafka消费者中查看相应数据。

#插入数据，sql
insert into student(id,name,age,grade)value(1,'Xiaoming',23,98);
insert into student(id,name,age,grade)value(2,'Zhangsan',24,96);
insert into student(id,name,age,grade)value(3,'Lisi',24,93);
insert into student(id,name,age,grade)value(4,'Wangwu',21,91);
insert into student(id,name,age,grade)value(5,'Weiliu',21,91);

解题

1.flume配置文件

# # 给sources（在一个agent里可以定义多个source）取个名字
a1.sources = r1
# # 给channel个名字
a1.channels = c1
# # 给channel个名字
a1.sinks = k1

#配置source
a1.sources.r1.type = org.keedio.flume.source.SQLSource
#ip地址和数据库名称需要修改，如果连接的本地的mysql改成本机IP地址，如果连接虚拟机上的mysql改成虚拟机的IP地址
a1.sources.r1.hibernate.connection.url = jdbc:mysql://ip地址/数据名称?useSSL=false&allowPublicKeyRetrieval=true
# mysql用户名，一般都是root
a1.sources.r1.hibernate.connection.user = root
# 密码
a1.sources.r1.hibernate.connection.password = 123456
a1.sources.r1.hibernate.connection.autocommit = true
# mysql驱动
a1.sources.r1.hibernate.dialect = org.hibernate.dialect.MySQL5Dialect
# 驱动版本过低会无法使用，驱动安装下文会提及
a1.sources.r1.hibernate.connection.driver_class = com.mysql.jdbc.Driver
a1.sources.r1.run.query.delay=5000

# 自定义查询
a1.sources.r1.start.from = 0
a1.sources.r1.table = student
a1.sources.r1.columns.to.select = *

a1.sources.r1.batch.size = 1000
a1.sources.r1.max.rows = 1000
a1.sources.r1.hibernate.connection.provider_class = org.hibernate.connection.C3P0ConnectionProvider
a1.sources.r1.hibernate.c3p0.min_size=1
a1.sources.r1.hibernate.c3p0.max_size=10
# 存放status文件
a1.sources.r1.status.file.path = /root/dir1
a1.sources.r1.status.file.name = r1.status
# 配置channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 10000
a1.channels.c1.byteCapacityBufferPercentage = 20
a1.channels.c1.byteCapacity = 800000
# 配置sinks
a1.sinks.k1.channel = c1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
# 指定主题名topic
a1.sinks.k1.kafka.topic = events
a1.sinks.k1.kafka.bootstrap.servers = master:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1

# 将source、channel、sink组装成agent
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

2.要保证配置中的文件路径存在，并且根据注释做出相应的修改

3.导入jar包到/flume/lib文件夹下

# flume连接mysqljar包
flume-ng-sql-source-1.4.3.jar
# mysql驱动jar包
mysql-connector-java-5.1.49.jar

5.flume-ng-sql-source项目的官方网址

https://github.com/keedio/flume-ng-sql-source

6.测试

1.创建mysql数据库、数据表
#创建数据库
create database school;
#使用数据库
use school;
#创建数据表
create table student(
	id int not null,
	name varchar(40),
	age int,
	grade int,
	primary key(id)
);
2.启动flume
flume-ng agent -n a1 -f spoolingtest.cong -Dflume.root.logger=DEBUG,console
3.启动kafka消费者
#启动消费者之前首先要创建主题
kafka-topics.sh --create --topic events --bootstrap-server master:9092
#启动消费者
kafka-console-consumer.sh --bootstrap-server master:9092 --topic events --from-beginning
4.student表中插入数据
#插入数据，sql
insert into student(id,name,age,grade)value(1,'Xiaoming',23,98);
insert into student(id,name,age,grade)value(2,'Zhangsan',24,96);
insert into student(id,name,age,grade)value(3,'Lisi',24,93);
insert into student(id,name,age,grade)value(4,'Wangwu',21,91);
insert into student(id,name,age,grade)value(5,'Weiliu',21,91);
5.观察数据

【flume实时采集mysql数据库的数据到kafka】

【flume实时采集mysql数据库的数据到kafka】
为了方便测试我重复插入了一条数据

6.遇到的问题：消费者没有消费出来数据，且flume配置文件没错
解决

删除存放的status文件，也就是我配置文件中/root/dir1文件夹中的东西
【flume实时采集mysql数据库的数据到kafka】

四、Kafka链接Flume，将生产的消息存入到HDFS

题目

编写配置文件，将kafka作为输入，在生产者中输入“HelloFlume”或其他信息，通过Flume将Kafka生产者输入的信息存入hdfs中，存储格式hdfs://localhost:9000/fromkafka/%Y%m%d/，要求存储时文件名为kafka_log。

解答

1.flume配置文件


agent.sources = kafka_source
agent.channels = mem_channel
agent.sinks = hdfs_sink
# 以下配置 source
agent.sources.kafka_source.type = org.apache.flume.source.kafka.KafkaSource
agent.sources.kafka_source.channels = mem_channel
agent.sources.kafka_source.batchSize = 5000
# master为主机名或者主机IP
agent.sources.kafka_source.kafka.bootstrap.servers = master:9092
# 需要指定kafka生产者的主题
agent.sources.kafka_source.kafka.topics = kafka-flume
# 以下配置 sink
agent.sinks.hdfs_sink.type = hdfs
agent.sinks.hdfs_sink.channel = mem_channel
# 输出到HDFS文件的路径
agent.sinks.hdfs_sink.hdfs.path = /data/kafka-flume/%Y%m%d
# HDFS文件前缀
agent.sinks.hdfs_sink.hdfs.filePrefix = kafka_log
# 使用时间戳
a1.sinks.r1.hdfs.useLocalTimeStamp = true

agent.sinks.hdfs_sink.hdfs.rollSize = 0
agent.sinks.hdfs_sink.hdfs.rollCount = 0
agent.sinks.hdfs_sink.hdfs.rollInterval = 3600
agent.sinks.hdfs_sink.hdfs.threadsPoolSize = 30
agent.sinks.hdfs_sink.hdfs.fileType=DataStream
agent.sinks.hdfs_sink.hdfs.writeFormat=Text
# 以下配置 channel
agent.channels.mem_channel.type = memory
agent.channels.mem_channel.capacity = 100000
agent.channels.mem_channel.transactionCapacity = 10000

2.根据配置文件的注释根据自己的情况做出修改

3.hadoop操作文件夹命令

# 创建文件夹
hadoop fs -mkdir 文件夹路径
# 删除文件夹
hadoop fs -rm -r -skipTrash 文件夹路径

4.测试

1.启动flume
#启动命令
flume-ng agent -n agent -f kafka-flume-hdfs.cong -Dflume.root.logger=DEBUG,console
# 参数解释
第二个agent为配置文件中agent的名字，kafka-flume-hdfs.cong为配置文件名，根据自己情况做出修改
2.启动kafka生产者
# 启动命令，根据自己配置变换主题名
kafka-console-producer.sh --broker-list master:9092 --topic kafka-flume
# 启动后，输入HelloFlume
3.观察HDFS界面，寻找并下载文件，查看内容

【flume实时采集mysql数据库的数据到kafka】

记得关闭flume后再下载，不然会下载出来.tmp文件文章来源地址https://www.toymoban.com/news/detail-437003.html

有什么问题评论区提问！

到了这里，关于【flume实时采集mysql数据库的数据到kafka】的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处： 如若内容造成侵权/违法违规/事实不符，请点击违法举报进行投诉反馈，一经查实，立即删除！

分享到：
领支付宝红包 赞助服务器费用

相关文章

  • Flume学习-采集端口数据存入kafka

    Flume学习-采集端口数据存入kafka

    启动zookeeper、kafka并创建kafka主题 2、创建flume-kafka.conf配置文件 用于采集socket数据后存入kafka 在flume文件夹中的conf下新建flume-kafka.conf配置文件 设置监听本地端口10050 netcat发送的socket数据，讲采集到的数据存入kafka的hunter主题中 3、启动flume ./bin/flume-ng ：启动Flume-ng二进制文件。

    2024年02月03日
    浏览(45)

  • 使用Flume-KafkaSource实时采集Avro格式数据

    Flume是一个可靠、可扩展且具有高可用性的分布式系统，用于在大规模数据集群中进行高效的日志聚合、收集和传输。Kafka是一个分布式流处理平台，用于处理高容量的实时数据流。本文将介绍如何使用Flume的KafkaSource来实时采集Avro格式的数据，并提供相应的源代码。 首先，确

    2024年02月07日
    浏览(42)

  • 实时Flink的数据库与Kafka集成优化案例

    在现代数据处理系统中，实时数据处理和分析是至关重要的。Apache Flink是一个流处理框架，可以用于实时数据处理和分析。在许多场景下，Flink需要与数据库和Kafka等消息系统进行集成，以实现更高效的数据处理。本文将讨论Flink与数据库和Kafka集成的优化案例，并提供实际示

    2024年02月20日
    浏览(41)
  • 【Kafka+Flume+Mysql+Spark】实现新闻话题实时统计分析系统（附源码）

    【Kafka+Flume+Mysql+Spark】实现新闻话题实时统计分析系统（附源码）

    需要源码请点赞关注收藏后评论区留言私信~~~ 新闻话题实时统计分析系统以搜狗实验室的用户查询日志为基础，模拟生成用户查询日志，通过Flume将日志进行实时采集、汇集，分析并进行存储。利用Spark Streaming实时统计分析前20名流量最高的新闻话题，并在前端页面实时显示

    2024年02月06日
    浏览(50)

  • Debezium系列之：使用Debezium2.X采集PostgreSQL15数据库到Kafka Topic，创建具有数据采集权限的账号

    需要使用Deb

    2024年02月12日
    浏览(50)
  • 【数仓】通过Flume+kafka采集日志数据存储到Hadoop

    【数仓】通过Flume+kafka采集日志数据存储到Hadoop

    【数仓】基本概念、知识普及、核心技术 【数仓】数据分层概念以及相关逻辑 【数仓】Hadoop软件安装及使用（集群配置） 【数仓】Hadoop集群配置常用参数说明 【数仓】zookeeper软件安装及集群配置 【数仓】kafka软件安装及集群配置 【数仓】flume软件安装及配置 【数仓】flum

    2024年03月17日
    浏览(58)
  • 大数据之使用Flume监听端口采集数据流到Kafka

    大数据之使用Flume监听端口采集数据流到Kafka

    前言 题目： 一、读题分析 二、处理过程   1.先在Kafka中创建符合题意的Kafka的topic  创建符合题意的Kafka的topic 2.写出Flume所需要的配置文件 3.启动脚本然后启动Flume监听端口数据并传到Kafka 启动flume指令 启动脚本，观察Flume和Kafka的变化 三、重难点分析 总结          本题

    2024年02月08日
    浏览(58)
  • 一百七十二、Flume——Flume采集Kafka数据写入HDFS中（亲测有效、附截图）

    一百七十二、Flume——Flume采集Kafka数据写入HDFS中（亲测有效、附截图）

    作为日志采集工具Flume，它在项目中最常见的就是采集Kafka中的数据然后写入HDFS或者HBase中，这里就是用flume采集Kafka的数据导入HDFS中 kafka_2.13-3.0.0.tgz hadoop-3.1.3.tar.gz apache-flume-1.9.0-bin.tar.gz # cd  /home/hurys/dc_env/flume190/conf # vi  evaluation.properties ### Name agent, source, channels and sink ali

    2024年02月09日
    浏览(48)
  • 日志采集传输框架之 Flume，将监听端口数据发送至Kafka

    日志采集传输框架之 Flume，将监听端口数据发送至Kafka

    1、简介                 Flume 是 Cloudera 提供的一个高可用的，高可靠的，分布式的海量日志采集、聚合和传 输的系统。Flume 基于流式架构，主要有以下几个部分组成。  主要组件介绍： 1）、 Flume Agent 是一个 JVM 进程，它以事件的形式将数据从源头送至目的。Agent 主

    2024年01月22日
    浏览(55)
  • （二十八）大数据实战——Flume数据采集之kafka数据生产与消费集成案例

    （二十八）大数据实战——Flume数据采集之kafka数据生产与消费集成案例

    本节内容我们主要介绍一下flume数据采集和kafka消息中间键的整合。通过flume监听nc端口的数据，将数据发送到kafka消息的first主题中，然后在通过flume消费kafka中的主题消息，将消费到的消息打印到控制台上。集成使用flume作为kafka的生产者和消费者。关于nc工具、flume以及kafka的

    2024年02月09日
    浏览(50)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包，优惠每天领

二维码1

领取红包

二维码2

领红包