实时数据处理

目录一、启动zookeeper三台同时二、启动kafka一台三、在kafka下创建topic四、配置flume脚本五、启动flume和flume脚本文件六、消费数据实时数据采集一、启动zookeeper三台同时进入zookeeper下bin目录cd /opt/module/zookeeper/bin/启动zookeeper./zkServer.sh start二、启动kafka一台进入Kafka目录下cd /opt/module/kafka/启动kafkabin/kafka-server-start.sh config/server.properties重新开启一台 master查看jpskafka是否启动成功三、在kafka下创建topickafka-topics.sh --create --bootstrap-server master:9092 --replication-factor 1 --partitions 4 --topic order【master是主节点名字。partitions后跟分区看具体给的分区数。最后order为topic名字】查看创建的topickafka-topics.sh --bootstrap-server master:9092 --list四、配置flume脚本进入flume中job目录下cd /opt/module/flume/job/在job目录下配置flume_kafka.conf文件vim flume_kafka.conf将下列内容添加至文件中### 1.#Name the components on this agent a1.sources r1 a1.sinks k1 a1.channels c1 ##### 2.#Describe/configure the source a1.sources.r1.type netcat a1.sources.r1.bind localhost ##【地址也可以主机名也可以】 a1.sources.r1.port 26001 ##【端口号具体看给的端口号】 ##### 3.#Describe the sink a1.sinks.k1.type org.apache.flume.sink.kafka.KafkaSink ##【flume采集数据到kafka如果是hdfs则需要修改】 a1.sinks.k1.kafka.bootstrap.servers master:9092 ##【9092端口号可以到server.properties配置文件里看】 a1.sinks.k1.kafka.topic order ##【order为kafka中topic的名字】 ##### 4.#Use a channel which buffers events in memory a1.channels.c1.type memory a1.channels.c1.capacity 1000 ##【flume通道内存量可以适当大一些防止数据溢出】 a1.channels.c1.transactionCapacity 100 ##【单次传数据容量】 # ##### 5.#Bind the source and sink to the channel # a1.sources.r1.channelsc1 a1.sinks.k1.channelc1 ##【绑定source、sink、channel】五、启动flume和flume脚本文件启动flume进入flume目录下cd /opt/module/flume/启动flume其中job/flume_kafka.conf 为自己新建脚本文件的路径bin/flume-ng agent --conf conf/ --name a1 --conf-file job/flume_kafka.conf -Dflume.root.loggerINFO,console再重新开启一台master启动flume脚本文件。cd /opt/mkdir jobcd job/再job目录下上传文件并给上传好的文件赋予权限chmod 777 socket_data_gen_update启动文件./socket_data_gen_update六、消费数据消费kafka中的数据(查看kafka中是否传入数据)再开启一台master进行消费数据【master、slave1、slave2是三台节点名字order是topic名字;--max-messages代表获取前几行数据】kafka-console-consumer.sh --bootstrap-server master:9092,slave1:9092,slave2:9092 --topic order --from-beginning --max-messages 10