利用FLume和Kafka实时读取数据日志数据操作手册

利用FLume和Kafka实时读取数据日志数据操作手册蜂箱数据的读取Flume可以实时采集网络数据并传输至本地，思考一下，脚本中所对应的Source、Channel和Sink应该如何配置。5G无线网络将传感器数据传输至本地，Flume是在本地进行接收，Source应该是应用于接收网络数据，数据类型应该配置为NetCat，这里只需要配置好IP地址和端口号即可。Channel用于数据缓存，由于是实时数据，这里使用存储速度较快的内存做为缓冲区。数据到达本地后要进行实时数据的处理，需要通过数据下沉至Kafka，通过KafKa再传递给Flink进行数据处理。Sink要设置的参数就是Kafka的接口。编写Flume脚本，读取端口数据，具体步骤如下。蜂箱数据采集Flume的脚本编写：cd/export/servers/flume-1.9.0/mkdirjobcdjobvikafka.conf在文件中添加如下内容：a1.sources=r1a1.sinks=k1a1.channels=c1a1.sources.r1.type=netcat#source负责接网络来的数据a1.sources.r1.bind=0#NetCatTCPSource监听的IP地址，传感器数据通过5G公网把数据发送本地，这里是模拟数据，生产环境是需要做公网地址与内网地址的转换a1.sources.r1.port=55566#监听的端口#kafkasinka1.sinks.k1.type=org.apache.flume.sink.kafka.KafkaSinka1.sinks.k1.kafka.bootstrap.servers=mater:9092#kafka服务的主机名为master，服务端口为9092a1.sinks.k1.kafka.topic=ordera1.sinks.k1.kafka.flumeBatchSize=20ducer.acks=1ducer.linger.ms=1#channel1a1.channels.c1.type=memorya1.sources.r1.channels=c1a1.sinks.k1.channel=c11.2.2启动脚本在/export/servers/flume-1.9.0/目录下执行命令bin/flume-ngagent--confconf/--namea1--conf-filebee.conf利用tel.sh模拟数据的发送在/bin目录下新建文件tel.sh，添加如下脚本：#!/bin/bashfunctionecho_sleep(){echo$1sleep2}#for((i=1;i<=1000;i++))while[TRUE];do(#sleep1echo_sleep"B区,一号蜂箱,25,0.6"echo_sleep"B区,二号蜂箱,25,0.65"echo_sleep"B区,三号蜂箱,20,0.55"echo_sleep"B区,一号蜂箱,25,0.6"echo_sleep"B区,二号蜂箱,25,0.65"echo_sleep"B区,二号蜂箱,25,0.65"echo_sleep"B区,三号蜂箱,20,0.65"echo_sleep"B区,三号蜂箱,20,0.55"echo_sleep"B区,二号蜂箱,25,0.6"echo_sleep"B区,二号蜂箱,25,0.65"echo_sleep"B区,一号蜂箱,25,0.6")|telnet055566echo"telnetfinish!"done修改脚本权限chmod777tel.sh在/bin目录下直接执行tel.sh即可。Kafka服务的配置Kafka在蜂箱数据中主要作用是将上游Flume采集到的数据，做为生产数据，再转发给下游的Flink进行消费。在要案例中只要把kafka服务启动，并创建对应的主题即可，但要注意的是kafka是依赖于zookeeper的，在配置kafka之前一定先启动zookeeper集群，启动方法详见，zookeeper的安装与配置。启动zookeeper后，就可以对启动kafkau并创建主题，具体操作步骤如下。(1)启动服务cd/export/servers/kafka/bin/kafka-server-start.shconfig/perties(2)创建主题cd/export/servers/kafka/bin/kafka-topics.sh--cr