大数据之Kafka———Flume日志系统写入kafka

我不是女神ヾ 2023-10-13 12:46 83阅读 0赞

### 一、核心原理 ###

![e2c60536073142a485dbe39ec090c8a4.png][]

1.通过Telnet 方式连接当前机器的端口

2.连接好端口可以在端口输入内容由 ng进行处理

3.启动ag 其中有三个组件Source(监管当前机器的端口)，Channel（进行分流），Sink（进行输出内容）

4.通过Sink将数据写到kb22在启动消费者

### 二、启动服务 ###

> 1.  zkServer.sh start
> 2.  nohup kafka-server-start.sh /opt/soft/kafka212/config/server.properties &

### 三、配置文件 ###

> 1.  \[root@kb144 ~\]\# cd /opt/soft/flume190/conf/myconf
> 2.  \[root@kb144 myconf\]\# vim ./netcat-flume-kafka.conf
>     
>     a1.sources = r1  
>     a1.sinks = k1  
>     a1.channels = c1
>     
>     a1.sources.r1.type=netcat  
>     a1.sources.r1.bind=localhost  
>     a1.sources.r1.port=7777
>     
>     a1.channels.c1.type = memory  
>     \# 4. 配置sink  
>     a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink  
>     a1.sinks.k1.brokerList=192.168.153.144:9092  
>     a1.sinks.k1.topic=kb22  
>     \# 5. 拼接组件  
>     a1.sources.r1.channels = c1  
>     a1.sinks.k1.channel = c1

### 四、启动kafka消费者 ###

> \[root@kb144 myconf\]\# kafka-console-consumer.sh --bootstrap-server 192.168.153.144:9092 --topic kb22

### 五、启动agent(flume的监听服务) ###

> \[root@kb144 conf\]\# ../bin/flume-ng agent --name a1 --conf . --conf-file ./myconf/netcat-flume-kafka.conf -Dflume.root.logger=INFO,console

### 六、通过telent方式接通flume 7777端口 ###

> （端口启动后可以在端口中输入内容 可以在上面启动的kafka里查看到）
> 
> \[root@kb144 myconf\]\# telnet localhost 7777
> 
> 如果端口启动不了查看一下端口是否被占用
> 
> netstat -lnp|grep 7777

### 七.把虚拟机存在的文件通过配置文件将其写入kafka中然后由消费者进行拉取 ###

1./opt/events/data文件下的users.csv

2.\[root@kb144 conf\]\# vim ./myconf/users.conf

> 配置文件
> 
> 定义组件  
> users.sources=usersSource  
> users.channels=usersChannel  
> users.sinks=usersSink
> 
> 配置Source
> 
> users.sources.usersSource.type=spooldir  
> users.sources.usersSource.spoolDir=/opt/events/flumelook/users  
> users.sources.usersSource.deserializer=LINE  
> users.sources.usersSource.deserializer.maxLineLength=320000  
> users.sources.usersSource.includePattern=users\_\[0-9\]\{4\}-\[0-9\]\{2\}-\[0-9\]\{2\}.csv  
> users.sources.usersSource.interceptors=head\_filter  
> users.sources.usersSource.interceptors.head\_filter.type=regex\_filter  
> users.sources.usersSource.interceptors.head\_filter.regex=^user\_id\*  
> users.sources.usersSource.interceptors.head\_filter.excludeEvents=true
> 
> 配置Channel
> 
> users.channels.usersChannel.type=file  
> users.channels.usersChannel.checkpointDir=/opt/events/checkpoint/users  
> users.channels.usersChannel.dataDir=/opt/events/dataDirs/users
> 
> 配置Sink
> 
> users.sinks.usersSink.type=org.apache.flume.sink.kafka.KafkaSink  
> users.sinks.usersSink.batchSize=640  
> users.sinks.usersSink.brokerList=192.168.153.144:9092  
> users.sinks.usersSink.topic=users
> 
> 拼接组件
> 
> users.sources.usersSource.channels=usersChannel  
> users.sinks.usersSink.channel=usersChannel  
>

3.创建队列 生产者

> \[root@kb144 data\]\# kafka-topics.sh --create --zookeeper 192.168.153.144:2181 --topic users --partitions 1 --replication-factor 1

4.启动agent

> \[root@kb144 conf\]\# ../bin/flume-ng agent --name users --conf . --conf-file ./myconf/users.conf -Dflume.root.logger=INFO,console

5.将文件拷贝到监控目录下

> \[root@kb144 data\]\#cp users.csv /opt/events/flumelook/users/users\_2023-07-06.csv

6.创建消费者

> \[root@kb144 kafka-logs\]\# kafka-console-consumer.sh --bootstrap-server 192.168.153.144:9092 --topic users --from-beginning

[e2c60536073142a485dbe39ec090c8a4.png]: https://img-blog.csdnimg.cn/e2c60536073142a485dbe39ec090c8a4.png