一、Flume

Flume主要由3个重要的组件构成：

Source:完成对日志数据的收集，分成transtion 和 event 打入到channel之中。
Channel:主要提供一个队列的功能，对source提供中的数据进行简单的缓存。
Sink:取出Channel中的数据，进行相应的存储文件系统，数据库，或者提交到远程服务器

Flume逻辑上分三层架构：agent，collector，storage

agent用于采集数据，agent是flume中产生数据流的地方，同时，agent会将产生的数据流传输到collector
collector的作用是将多个agent的数据汇总后，加载到storage中。
storage是存储系统，可以是一个普通file，也可以是HDFS，HIVE，HBase等。

关于Flume更多内容，可以参考网络文献：Flume的原理和使用

Flume-Kafka-HBase的集成

在/usr/local/flume/conf下面配置flume-conf.properties文件（这个文件可以由flume-conf.properties.template复制得到）

agent1.sources = r1
agent1.channels = kafkaC hbaseC
agent1.sinks = kafkaSink hbaseSink

agent1.sources.r1.type = exec
agent1.sources.r1.command = tail -F  /opt/datas/weblogs-flume.log
agent1.sources.r1.channels = kafkaC hbaseC

#*****************flume+Kafka***********************
agent1.channels.kafkaC.type = memory
agent1.channels.kafkaC.capacity = 100000
agent1.channels.kafkaC.transactionCapacity = 100000
agent1.channels.kafkaC.keep-alive = 20

agent1.sinks.kafkaSink.channel = kafkaC
agent1.sinks.kafkaSink.type = org.apache.flume.sink.kafka.KafkaSink
agent1.sinks.kafkaSink.brokerList = localhost:9092
agent1.sinks.kafkaSink.topic = weblogs
agent1.sinks.kafkaSink.zookeeperConnect = localhost:2181
agent1.sinks.kafkaSink.requiredAcks = 1
agent1.sinks.kafkaSink.batchSize = 1


agent1.channels.hbaseC.type = memory
agent1.channels.hbaseC.capacity = 100000
agent1.channels.hbaseC.transactionCapacity = 100000
agent1.channels.hbaseC.keep-alive = 20

agent1.sinks.hbaseSink.type = hbase
agent1.sinks.hbaseSink.table = weblogs
agent1.sinks.hbaseSink.columnFamily = info
agent1.sinks.hbaseSink.serializer = org.apache.flume.sink.hbase.RegexHbaseEventSerializer
# 这个正则表达式临时之举，不过证明确实好用
agent1.sinks.hbaseSink.serializer.regex = (.*?),(.*?),\\[(.*?)\\],(.*?),(.*?),(.*)
agent1.sinks.hbaseSink.serializer.colNames = datatime,userid,searchname,retorder,cliorder,cliurl
agent1.sinks.hbaseSink.channel = hbaseC

需要在Hbase中创建相关的表

hbase shell
create 'weblogs','info'

编写flume的启动脚本

vi flume-start.sh

#/bin/bash

echo "flume start"

# 这里的agent的名字必须与flume-conf.properties中的agent保持一致
flume-ng agent --conf conf -f /usr/local/flume/conf/flume-conf.properties -n agent1 -Dflume.root.logger=INFO,console

二、启动服务

如果显示找不到相关命令，说明需要修改环境变量

启动Zookeeper服务

zkServer.sh start

2. 启动hdfs服务

start-dfs.sh

3. 启动HBase服务

start-hbase.sh

4. 启动Kafka服务

# & 放在启动参数后面表示设置此进程为后台进程，进程切换到后台的时候，我们把它称为job
kafka-server-start.sh /usr/local/kafaka/config/server.properties &

5. 创建业务数据topic

kafka-topics.sh --zookeeper localhost:2181 --create --topic weblogs --replication-factor 1 --partitions 1

6. 启动模拟日志生成脚本

# cd 到你脚本所在的文件
./weblog-shell.sh

7. 启动flume脚本

./flume-collect-start.sh

8. 启动Kafka Consumer查看flume日志采集情况

kafka-console-consumer.sh --zookeeper localhost:2181 --topic weblogs --from-beginning

9. 查看hbase数据写入情况

./hbase-shell
count 'weblogs'

ITPub博客

flume和Kafka、HBase的集成

一、Flume

Flume-Kafka-HBase的集成

二、启动服务

最新文章