搭建flume+kafka+storm環境

這裡，我的flume是採集mysql的資料再存入kafka，我用mysql作為我的source，記憶體memory作為channels，kafka作為sink，這個要藉助一個外掛source-ng-sql手機mysql的資料

環境準備：
1.flume1.8.0
2.kafka2.0.0
3.storm1.2.2
4.zookeeper3.4.13

flume
下載：
http://flume.apache.org/download.html
下載
apache-flume-1.8.0-bin.tar.gz
解壓
tar -zxvf ~/Desktop/apache-flume-1.8.0-bin.tar.gz
sudo mv apache-flume-1.8.0-bin flume1.8.0
mkdir -p bigdata
sudo mv flume1.8.0 /usr/local/bigdata
改一下擁有者
sudo chown bigdata -R storm：storm
修改flume的conf配置檔案
sudo cp flume-conf.properties.template flume.conf
sudo vim flume.conf

a1.sources=sql
a1.channels=mem
a1.sinks=kafka

1.sources.sql.type=org.keedio.flume.source.SQLSource
a1.sources.sql.hibernate.connection.url=jdbc:mysql://資料庫地址

a1.sources.sql.hibernate.connection.user=root
a1.sources.sql.hibernate.connection.password=123456
a1.sources.sql.hibernate.connection.autocommit=
 
true
a1.sources.sql.hibernate.dialect=org.hibernate.dialect.MySQL5Dialect
a1.sources.sql.hibernate.connection.driver_class=com.mysql.jdbc.Driver
a1.sources.sql.run.query.delay=5000
a1.sources.sql.status.file.path=sql狀態檔案儲存路徑
a1.sources.sql.status.file.name=sql狀態檔名
# Custom query
a1.sources.sql.start.from
 
=0
a1.sources.sql.custom.query=資料庫查詢語句
a1.sources.sql.batch.size=1000
a1.sources.sql.max.rows=1000
a1.sources.sql.hibernate.connection.provider_class=org.hibernate.connection.C3P0ConnectionProvider
a1.sources.sql.hibernate.c3p0.min_size=1
a1.sources.sql.hibernate.c3p0.max_size=10
# use memory as the channels
a1.channels.mem.type=memory
a1.channels.mem.capacity=1000
a1.channels.mem.transactionCapacity=1000
a1.channels.mem.byteCapacityBufferPercentage=20
a1.channels.mem.byteCapacity=800000

a1.sinks.kafka.type=org.apache.flume.sink.kafka.KafkaSink
a1.sinks.kafka.topic=建立的kafka topic名
a1.sinks.kafka.brokerList=master:9092,slave1:9092,slave2:9092
a1.sinks.kafka.requireAcks=1
a1.sinks.kafka.batchSize=20
a1.sinks.kafka.partitionIdHeader=0

a1.sources.sql.channels=mem
a1.sinks.kafka.channel=mem

配置環境變數
sudo vim /etc/profile

flume-ng-sql-source外掛
下載
https://github.com/xiaomoo/oeasy
在flume的目錄下建立目錄plugins.d

目錄結構如上
lib放編譯好的flume-ng-sql-source.jar
libext放mysql的jar

zookeeper
下載：
http://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.4.13/
解壓：
tar -zxvf zookeeper-3.4.13.tar.gz
sudo mv zookeeper-3.4.13 zookeeper3.4.13
sudo zookeeper3.4.13 /usr/local/bigdata
修改配置檔案
sudo vim zoo.cfg
單節點配置

# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial 
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between 
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just 
# example sakes.
dataLogDir=/usr/local/bigdata/zookeeper3.4.13/logs
dataDir=/usr/local/bigdata/zookeeper3.4.13/data
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
maxClientCnxns=60
#
# Be sure to read the maintenance section of the 
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
#autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature
#autopurge.purgeInterval=1

然後新建data目錄，單節點為standalone模式

zookeeper叢集配置

# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial 
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between 
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just 
# example sakes.
dataLogDir=/usr/local/bigdata/zookeeper3.4.13/logs
dataDir=/usr/local/bigdata/zookeeper3.4.13/data
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
maxClientCnxns=60
#
# Be sure to read the maintenance section of the 
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
#autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature
#autopurge.purgeInterval=1
server.1=0.0.0.0:2888:3888
server.2=slave1:2888:3888
server.3=slave2:2888:3888

新建data目錄
mkdir -p data
cat 1 > myid
通過ssh將zookeeper目錄傳送給slave1，slave2
scp -t zookeeper3.4.13 [email protected]:/usr/local/bigdata/
scp -t zookeeper3.4.13 [email protected]:/usr/local/bigdata/
分別將slave1，slave2新建data目錄
slave1：
cat 2 > myid
修改zoo.cfg

server.1=master:2888:3888
server.2=0.0.0.0:2888:3888
server.3=slave2:2888:3888

slave2：
cat 3 > myid
修改zoo.cfg

server.1=master:2888:3888
server.2=slave1:2888:3888
server.3=0.0.0.0:2888:3888

zookeeper叢集會選擇一個leader，其他為follower
配置環境變數

通過bin/zkServer.sh start啟動zookeeper
bin/zkServer.sh stop關閉zookeeper
bin/zkServer.sh status檢視叢集狀態

kafka
下載：
http://kafka.apache.org/downloads
解壓：
tar -zxvf kafka_2.11-2.0.0.tgz
sudo mv kafka_2.11-2.0.0 kafka2.0.0
sudo mv kafka2.0.0 /usr/local/bigdata
cd /usr/local/bigdata/kafka2.0.0/config
修改kafka配置server.properties
sudo vim server.properties

broker.id=0
host.name=本機的ip地址

listeners=PLAINTEXT://192.168.5.211:9092
log.dirs=/usr/local/bigdata/kafka2.0.0/logs

num.network.threads=3
num.io.threads=8

socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400

socket.request.max.bytes=104857600
delete.topic.enable=true

log.dirs=/usr/local/bigdata/kafka2.0.0/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1

offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

zookeeper.connect=你的zookeeper地址，可以使單節點也可以是叢集
zookeeper.connection.timeout.ms=6000
roup.initial.rebalance.delay.ms=0

開啟zookeeper.properties
sudo vim zookeeper.properties

dataDir=你的zookeeper的data目錄路徑
clientPort=2181

將kafka傳送到slave1，slave2
scp -t kafka2.0.0 [email protected]:/usr/local/bigdata/
scp -t kafka2.0.0 [email protected]:/usr/local/bigdata/
分別修改slave1，slave2的kafka中config的server.properties
slave1：
將broker.id改為1
slave2：
將broker.id改為2

配置環境變數

storm
下載
http://storm.apache.org/downloads.html
解壓
tar -zxvf apache-storm-1.2.2.tar.gz
sudo mv apache-storm-1.2.2 storm1.2.2
sudo mv storm1.2.2 /usr/local/bigdata

修改配置檔案,在conf目錄下
sudo vim storm.yaml

storm.local.dir: "/usr/local/bigdata/storm1.2.2/localdir"
storm.zookeeper.servers:
     - zookeeper地址
     
nimbus.seeds: ["master"]
ui.host: 0.0.0.0
ui.port: 8093
supervisor.slots.ports:
     - 6700
     - 6701
     - 6702
     - 6703

drpc.servers:
     - "master"
     - "slave1"
     - "slave2"

將storm傳送給slave1，slave2
scp -t storm1.2.2 [email protected]:/usr/local/bigdata/
scp -t storm1.2.2 [email protected]:/usr/local/bigdata/

配置環境變數

測試環境
首先啟動zookeeper
$ZK_HOME/bin/zkServer.sh start(我用的單節點)
$ZK_HOME/bin/zkServer.sh status檢視狀態

啟動kafka叢集
$KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties &

啟動storm
master：
storm nimbus &
storm ui &

slave1:
storm supervisor &

slave2:
storm supervisor &

jps命令檢視啟動程序
master：

slave1：

slave2：

至此，環境就搭建完成了。