TB级大数据集群日志采集安装配置详细

[原创] TB级大数据集群日志采集安装配置详细

仧尐 2018/12/19 17 flume/ kafka

目前，市面上日志采集一般使用Flume和Kafka组件。搭建集群一般需要至少准备三台机器，比如我使用的荣耀路由器的网段是：192.168.3.*。暂且将虚拟机的IP设置为：192.168.3.101,192.168.3.102,192.168.3.103 别名分别为：server1,server2,server3

在每台机器上修改hosts文件：

```shell
vi /etc/hosts
#写入集群主机名映射关系
192.168.3.101    server1
192.168.3.102    server2
192.168.3.103    server3
```

目前先在server1做如下步骤，后面再分发到server2,server3。

### 第一步：安装JDK

到oracle官网下载[jdk-8u101-linux-x64.tar.gz](http://download.oracle.com/otn-pub/java/jdk/8u101-b14/jdk-8u101-linux-x64.tar.gz?AuthParam=1483251511_04673c3489fc71bc4768a7cc206b3ae4)

①先查找并卸载自带openJDK

```
rpm -qa |grep openjdk
yum -y remove *openjdk*
```

②解压[jdk-8u101-linux-x64.tar.gz](http://download.oracle.com/otn-pub/java/jdk/8u101-b14/jdk-8u101-linux-x64.tar.gz?AuthParam=1483251511_04673c3489fc71bc4768a7cc206b3ae4)到安装目录，如果没有特别说明，全文中安装目录都是指/data/soft/

```shell
remove jdk-8u101-linux-x64.tar.gz /data/soft/
cd /data/soft

tar zxvf jdk-8u101-linux-x64.tar.gz
mv jdk-8u101-linux-x64.tar.gz java
```

③配置JAVA_HOME

```shell
vi ~/.bashrc
#在文件的末端写入如下配置
## java configuration
export JAVA_HOME=/data/soft/java
export PATH=.:$JAVA_HOME/bin:$PATH
#使配置生效
source ~/.bashrc
```

④验证是否安装成功

```
java -version
```

### 第二步：Zookeeper安装

#### 2.1 Zookeeper下载准备

去Zookeeper官网或者用wget下载安装包[zookeeper-3.4.9.tar.gz](http://mirrors.cnnic.cn/apache/zookeeper/zookeeper-3.4.9/zookeeper-3.4.9.tar.gz)

#### 2.2 Zookeeper安装配置

①把[zookeeper-3.4.9.tar.gz](http://mirrors.cnnic.cn/apache/zookeeper/zookeeper-3.4.9/zookeeper-3.4.9.tar.gz)移动到安装目录并解压

```shell
mv zookeeper-3.4.9.tar.gz /data/soft/
cd /data/soft 
tar -zxvf zookeeper-3.4.9.tar.gz 
mv zookeeper-3.4.9 zookeeper
```

②编辑并保存zoo.cfg，配置文件如下

```shell
vi /data/soft/zookeeper/conf/zoo.cfg
#写入以下配置
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data/soft/zookeeper
clientPort=2181
server.1=server1:2888:3999
server.2=server2:2888:3999
server.3=server3:2888:3999
```

③在dataDir下新增文件myid，填写值为server.*对应的编号

```shell
#如果是server1，则写1，server2，则写2，server3，则写3
echo '1'> /data/soft/zookeeper/myid
```

④添加zookeeper环境变量

```shell
vi ~/.bashrc
#写入一下内容
## zk configuration
export ZOOKEEPER_HOME=/data/soft/zookeeper
export PATH=$ZOOKEEPER_HOME/bin:$PATH
#使上面的配置生效
source ~/.bashrc
```

⑤启动zookeeper并查看状态，可以看到节点是leader还是follower

```shell
#启动当前节点
zkServer.sh start
#查看当前节点的状态
zkServer.sh status
```

### 第三步：Kafka安装

#### 3.1 Kafka下载准备

从Kafka官网或者用wget下载Kafka安装包[kafka_2.10-0.10.1.1.tgz](http://mirror.bit.edu.cn/apache/kafka/0.10.1.1/kafka_2.10-0.10.1.1.tgz)

#### 3.2 Kafka安装配置

①移动并解压到安装目录/data/soft

```shell
mv kafka_2.10-0.10.1.1.tgz /data/soft/
cd /data/soft
tar -zxvf kafka_2.10-0.10.1.1.tgz
mv kafka_2.10-0.10.1.1 kafka
```

②修改配置文件

```shell
vi /data/soft/kafka/config/server.properties
#替换为
broker.id=1
advertised.host.name=server1
host.name=server1
auto.create.topics.enable=true
delete.topic.enable=true
default.replication.factor=3
num.network.threads=3
num.io.threads=8
num.partitions=3
num.recovery.threads.per.data.dir=1
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
delete.topic.enable=true
log.dirs=/data/soft/kafka/logs
#过期时间 设置7天过期
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
listeners=PLAINTEXT://:9092
zookeeper.connection.timeout.ms=6000
zookeeper.connect=server1:2181,server2:2181,server3:2181
```

③创建日志目录，并赋予读写权限

```shell
mkdir -p /data/soft/kafka/logs
chmod -R 777 /data/soft/kafka/logs 
```

④添加kafka环境变量

```shell
vi ~/.bashrc
#在文件中
## zk configuration 
export KAFKA_HOME=/data/soft/kafka
export PATH=$KAFKA_HOME/bin:$PATH
source ~/.bashrc
```

⑤修改kafka内存设置

```shell
vi kafka-server-start.sh
#在头部插入，如果内存充足可以上调
export KAFKA_HEAP_OPTS="-Xmx3G -Xms1G"
#wq
```

⑥启动kafka并查看状态

```shell
kafka-server-start.sh -daemon  /data/soft/kafka/config/server.properties
kafka-console-agent.sh --broker-list server1:9092,server2:9092,server3:9092 --topic demo_test
kafka-topics.sh --list --zookeeper server1:2181,server2:2181,server3:2181
```

### 第四步：Flume安装

#### 4.1 Flume下载准备

去Flume官网或者用wget下载安装包[apache-flume-1.7.0-bin.tar.gz](http://www.apache.org/dyn/closer.lua/flume/1.7.0/apache-flume-1.7.0-bin.tar.gz)

如果有修改源码的需求，可以下载源码包：[apache-flume-1.7.0-src.tar.gz](http://www.apache.org/dyn/closer.lua/flume/1.7.0/apache-flume-1.7.0-src.tar.gz)

Flume需要依赖jdk1.7以上，这里不再描述。

#### 4.2Flume安装与配置

①把[apache-flume-1.7.0-bin.tar.gz](http://www.apache.org/dyn/closer.lua/flume/1.7.0/apache-flume-1.7.0-bin.tar.gz)移动到待安装目录并解压

```shell
mv apache-flume-1.7.0-bin.tar.gz /data/soft
cd /data/soft
tar -zxvf apache-flume-1.7.0-bin.tar.gz
mv apache-flume-1.7.0-bin flume
```

②配置好FLUME_HOME的环境变量

```shell
vi ~/.bashrc
## flume configuration
export FLUME_HOME=/data/soft/flume
export PATH=$FLUME_HOME/bin:$PATH
source ~/.bashrc
```

③查看版本，如果显示版本正常则表示安装成功。

```shell
#查看版本验证安装是否成功
flume-ng version
```

④配置(从文件=>Kafka通道)

```shell
#创建两个从文件夹里读取文件下沉到Kafka的通道 名为：xx1 xx2
mkdir -p /data/path/xx1
mkdir -p /data/path/xx2
cd /data/soft/flume/conf/
#create properties
vi collect-conf.properties
#下面是配置内容：

##########################################
# agent config
###########################################

#*****************agent section**********************
agent.sources = source_xx1 source_xx2
agent.channels = channel_xx1 channel_xx2
agent.sinks = sink_xx1 sink_xx2

#*****************source section**********************
#xx1 source section
#spooldir:监控一个目录下的文件
agent.sources.source_xx1.type = spooldir
#删除策略 文件处理完立即删除，不再保留
agent.sources.source_xx1.deletePolicy=immediate
agent.sources.source_xx1.channels = channel_xx1
agent.sources.source_xx1.fileHeader = true
agent.sources.source_xx1.fileHeaderKey=xx1
agent.sources.source_xx1.spoolDir=/data/path/xx1
agent.sources.source_xx1.batchSize=500
agent.sources.source_xx1.bufferMaxLineLength=20000
agent.sources.source_xx1.inputCharset=UTF-8
agent.sources.source_xx1.decodeErrorPolicy=IGNORE
#自定义拦截器
agent.sources.source_xx1.interceptors=i1
agent.sources.source_xx1.interceptors.i1.type=xx.xx.xx.xx.XxxxInterceptor

#xx2 source section
agent.sources.source_xx2.type = spooldir
agent.sources.source_xx2.deletePolicy=immediate
agent.sources.source_xx2.channels = channel_xx2
agent.sources.source_xx2.fileHeader = true
agent.sources.source_xx2.fileHeaderKey=xx2
agent.sources.source_xx2.spoolDir=/data/path/xx2
agent.sources.source_xx2.batchSize=500
agent.sources.source_xx2.bufferMaxLineLength=20000
agent.sources.source_xx2.inputCharset=UTF-8
agent.sources.source_xx2.decodeErrorPolicy=IGNORE

#*****************channel section**********************
#xx1 channel section
agent.channels.channel_xx1.type = memory
agent.channels.channel_xx1.capacity = 10000
agent.channels.channel_xx1.transactionCapacity=1000

#xx2 channel section
agent.channels.channel_xx2.type = memory
agent.channels.channel_xx2.capacity = 10000
agent.channels.channel_xx2.transactionCapacity=1000

#*****************sink section**********************
#xx1 sink section
agent.sinks.sink_xx1.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.sink_xx1.kafka.bootstrap.servers=collector1:9092,collector2:9092,collector3:9092
agent.sinks.sink_xx1.kafka.flumeBatchSize=1000
#1 one,0 none,-1 all
agent.sinks.sink_xx1.kafka.agent.acks=1
agent.sinks.sink_xx1.kafka.agent.type=sync
agent.sinks.sink_xx1.kafka.topic=kafka_topic_xx1
agent.sinks.sink_xx1.kafka.agent.compression.type = snappy
agent.sinks.sink_xx1.kafka.agent.linger.ms=1
agent.sinks.sink_xx1.channel = channel_xx1

#xx2 sink section
agent.sinks.sink_xx2.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.sink_xx2.kafka.bootstrap.servers=collector1:9092,collector2:9092,collector3:9092
agent.sinks.sink_xx2.kafka.flumeBatchSize=200
#1 one,0 none,-1 all
agent.sinks.sink_xx2.kafka.agent.acks=1
agent.sinks.sink_xx2.kafka.agent.type=sync
agent.sinks.sink_xx2.kafka.topic=kafka_topic_xx2
agent.sinks.sink_xx2.kafka.agent.compression.type = snappy
agent.sinks.sink_xx2.kafka.agent.linger.ms=1
agent.sinks.sink_xx2.channel = channel_xx2
```

⑤配置（从Kafka=>hdfs[或文件]通道）

```shell
cd /data/soft/flume/conf/
#create properties
vi collect-conf.properties
#下面是配置内容：

##########################################
# agent config
###########################################

#*****************agent section**********************
agent.sources = source_xx1
agent.channels = channel_xx1
agent.sinks = sink_xx1

#*****************source section**********************
agent.sources.source_xx1.type=org.apache.flume.source.kafka.KafkaSource
agent.sources.source_xx1.kafka.bootstrap.servers=server1:9092,server2:9092,server3:9092
agent.sources.source_xx1.kafka.topics=kafka_topic_xx1
agent.sources.source_xx1.channels=channel_db_data

#*****************channel section**********************
agent.channels.channel_xx1.type = memory
agent.channels.channel_xx1.capacity = 10000
agent.channels.channel_xx1.keep-alive=60
agent.channels.channel_xx1.transactionCapacity=1000

#*****************sink section**********************
#下沉到hdfs文件
agent.sinks.sink_xx1.type=hdfs
agent.sinks.sink_xx1.hdfs.path=hdfs://hdfs集群别名/data/xx1/%{day}
#一个小时创建一个新文件的策略
agent.sinks.sink_xx1.hdfs.rollInterval=3600
#值为0表示不按照文件的大小来滚动文件 
agent.sinks.sink_xx1.hdfs.rollSize=0
#值为0表示不按照events的条目来滚动文件
agent.sinks.sink_xx1.hdfs.rollCount=0
#如果一个文件4800秒没有进行操作则关闭
agent.sinks.sink_xx1.hdfs.idleTimeout=4800
agent.sinks.sink_xx1.hdfs.writeFormat=Text
#hdfs超时时间60秒
agent.sinks.sink_xx1.hdfs.callTimeout=60000
#文件写完后的后缀 变量day可以在events中的headers中设置，见自定义
agent.sinks.sink_xx1.hdfs.filePrefix=yourpre.%{day}
#正在写的文件的后缀
agent.sinks.sink_xx1.hdfs.inUseSuffix=.tmp
agent.sinks.sink_xx1.hdfs.batchSize=1000
agent.sinks.sink_xx1.hdfs.fileType=DataStream
agent.sinks.sink_xx1.channel=channel_xx1

#可选下沉到文件系统
#*****************sink section**********************
#agent.sinks.sink_xx1.type=file_roll
#agent.sinks.sink_xx1.chanel=chanel_xx1
#agent.sinks.sink_xx1.sink.directory=/data/xx1/%{day}
#agent.sinks.sink_xx1.sink.pathManager=DEFAULT
#agent.sinks.sink_xx1.sink.pathManager.extension=log
#一个小时滚动一次
#agent.sinks.sink_xx1.sink.rollInterval=3600
#agent.sinks.sink_xx1.sink.serializer=TEXT
#agent.sinks.sink_xx1.batchSize=100
```

⑥配置（下沉到flume->flume->hdfs（或文件））

```shell
#同一个chanel配置多个sink，因为可能有多个flume节点
agent.sources = source_xx
agent.channels = channel_xx
#将chanel_xx下沉到sink_xx1和sink_xx2
agent.sinks = sink_xx1 sink_xx2

#将数据下沉到192.168.3.104上的flume，使用44446端口接受
agent.sinks.sink_xx1.type=avro
agent.sinks.sink_xx1.hostname=192.168.3.104上的flume，使用
agent.sinks.sink_xx1.port=44446
agent.sinks.sink_xx1.threads=10
agent.sinks.sink_xx1.channel=channel_xx

#将数据下沉到192.168.3.105上的flume，使用44446端口接受
agent.sinks.sink_xx2.type=avro
agent.sinks.sink_xx2.hostname=192.168.3.105
agent.sinks.sink_xx2.port=44446
agent.sinks.sink_xx2.threads=10
agent.sinks.sink_xx2.channel=channel_xx

#两个sink进行负载均衡
agent.sinkgroups=sink_xx_group
agent.sinkgroups.gadx.sinks=sink_xx1 sink_xx2
agent.sinkgroups.gadx.processor.type = load_balance
agent.sinkgroups.gadx.processor.selector = round_robin
agent.sinkgroups.gadx.processor.backoff = true
#======================================================
#192.168.3.104和192.168.3.105上的flume配置
#*****************agent section**********************
agent.sources=source_aaa
agent.channels=chanel_aaa
agent.sinks=sink_aaa

#source配置
agent.sources.source_pad.type = avro
agent.sources.source_pad.channels = channel_pad
agent.sources.source_pad.bind=0.0.0.0
#监听44446端口
agent.sources.source_pad.port=44446

#*****************channel section**********************
agent.channels.channel_aaa.type = memory
agent.channels.channel_aaa.capacity = 10000
agent.channels.channel_aaa.keep-alive=60
agent.channels.channel_aaa.transactionCapacity=1000

#*****************sink section**********************
#下沉到hdfs文件
agent.sinks.sink_aaa.type=hdfs
agent.sinks.sink_aaa.hdfs.path=hdfs://hdfs集群别名/data/aaa/%{day}
#一个小时创建一个新文件的策略
agent.sinks.sink_aaa.hdfs.rollInterval=3600
#值为0表示不按照文件的大小来滚动文件 
agent.sinks.sink_aaa.hdfs.rollSize=0
#值为0表示不按照events的条目来滚动文件
agent.sinks.sink_aaa.hdfs.rollCount=0
#如果一个文件4800秒没有进行操作则关闭
agent.sinks.sink_aaa.hdfs.idleTimeout=4800
agent.sinks.sink_aaa.hdfs.writeFormat=Text
#hdfs超时时间60秒
agent.sinks.sink_aaa.hdfs.callTimeout=60000
#文件写完后的后缀 变量day可以在events中的headers中设置，见自定义
agent.sinks.sink_aaa.hdfs.filePrefix=yourpre.%{day}
#正在写的文件的后缀
agent.sinks.sink_aaa.hdfs.inUseSuffix=.tmp
agent.sinks.sink_aaa.hdfs.batchSize=1000
agent.sinks.sink_aaa.hdfs.fileType=DataStream
agent.sinks.sink_aaa.channel=channel_aaa
```

由于flume默认配置jvm的内存太小，这里修改一下启动参数，找到JAVA_OPTS这一行

```shell
vi bin/flume-ng
JAVA_OPTS="-Xms512m -Xmx3072m"
```

后台启动Flume，开启http监控

```shell
#其中-n 表示配置文件的key的前缀agent，也就是agent的名称。
nohup flume-ng agent -c /data/soft/flume/conf -f /data/soft/flume/conf/collect-conf.properties -n agent -Dflume.monitoring.type=http -Dflume.monitoring.port=34545 > /data/soft/flume/logs/cat.out 2>&1 & ps -ef|grep flume
```

访问监听端口[http://IP:34545](http://IP:34545)，可以看到返回的json状态

### 第五步：软件配置分发及验证

如果有100台机器都要一台一台配置那是会崩溃的

#### 5.1 Java分发

在主节点执行分发命令

```shell
for i in {2..3};do scp -r /data/soft/java/ root@server$i:/data/soft;done;
```

#### 5.2 Zookeeper分发

在主节点执行分发命令

```shell
for i in {2..3};do scp -r /data/soft/zookeeper/ root@server$i:/data/soft;done;
```

在各从节点修改myid，对照zoo.cfg中server.*对应的编号，比如该节点IP对应的server号码是2，则

```shell
echo '2' > /data/soft/zookeeper/myid
#依次类推
```

#### 5.3 Kafka分发

在主节点执行分发命令

```shell
for i in {2..3};do scp -r /data/soft/kafka/ root@server$i:/data/soft;done;
```

在各节点上修改broker.id

```shell
vi /data/soft/kafka/config/server.properties
#将broker.id替换为，数字随节点递增即可，每个节点要求不一致
broker.id=2
```

#### 5.4 Flume分发

在主节点执行分发命令

```shell
for i in {2..3};do scp -r /data/soft/flume/ root@server$i:/data/soft;done;
```

在各从节点上创建必要的数据目录，要求运行用户有读写权限

```shell
mkdir -p /data/path/xx1
chmod 775 /data/path/xx1
```

修改collect-conf.properties中的agent名称，即配置文件中第一个前缀，注意：启动时参数中agent -n要与配置文件一致

#### 5.5 环境变量配置分发

最后本次安装配置的.bashrc文件环境变量最终合并为以下

```shell
## java configuration
export JAVA_HOME=/data/soft/java
## zk configuration
export ZOOKEEPER_HOME=/data/soft/zookeeper
## flume configuration
export FLUME_HOME=/data/soft/flume
## zk configuration
export KAFKA_HOME=/data/soft/kafka
export PATH=$JAVA_HOME/bin:$ZOOKEEPER_HOME/bin:$KAFKA_HOME/bin:$FLUME_HOME/bin:$PATH
```

修改后，分发.bashrc文件并使之生效

```shell
for i in {2..3};do scp -r ~/.bashrc root@server$i:~/.bashrc;done;
#在各节点执行
source ~/.bashrc
```

#### 5.6 启动验证节点

#####  5.6.1 java验证

在各从节点上执行验证

```shell
java -version
#是否显示正确版本
```

##### 5.6.2 Zookeeper启动验证

在各从节点上启动zookeeper并查看状态，可以看到节点是leader还是follower

```shell
zkServer.sh start
zkServer.sh status
```

##### 5.6.3 Kafka启动验证

在从节点启动kafka并查看状态

```shell
kafka-server-start.sh -daemon  /data/soft/kafka/config/server.properties
#以下命令在集群节点执行即可，不限于同一个节点
kafka-console-producer.sh --broker-list server1:9092,server2:9092,server3:9092 --topic demo_test
kafka-topics.sh --list --zookeeper server1:2181,server2:2181,server3:2181 
```

##### 5.6.4 Flume启动验证

后台启动flume，开启http监控，注意：各节点agent名称最好配置成不一致的，以后方便集群，命令行的参数要与对应配置文件中前缀一致，如：-n agent

```shell
nohup flume-ng agent -c /data/soft/flume/conf -f /data/soft/flume/conf/collect-conf.properties -n agent -Dflume.monitoring.type=http -Dflume.monitoring.port=34545 > /data/soft/flume/logs/cat.out 2>&1 & ps -ef|grep flume
```

访问监听端口[http://IP:34545](http://IP:34545)，可以看到返回的json状态

验证zookeeper、kafka、flume整合是否成功：

使用mv命令(cp命令不建议使用)文本到flume指定文件夹/data/path/xx1，此时应该用命令可以读取到数据

```shell
kafka-console-consumer.sh --zookeeper server1:2181,server2:2181,server3:2181  --from-beginning --topic kafka_topic_xx1
```

### 附：Flume拦截器实现

案例：在采集日志的过程，如果下沉数据到hdfs文件，一般来讲都是存进hive表的分区文件夹中，那么我们怎么实现将数据完完整整地路由到当天的分区中呢？可以根据日志中的请求时间戳字段来进行路由。首先我们需要实现一个flume拦截器，且在每个事件的Headers中增加一个key为day，值为请求时间所在的日期。该key可以在配置文件中使用，如：

```shell
agent.sinks.sink_aaa.hdfs.path=hdfs://hdfs集群别名/data/aaa/%{day}
```

拦截器实现如下：

```java
package org.apache.flume.interceptor;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 编写该拦截器，需要引用flume的jar包
**/
public class TimestampInterceptor implements Interceptor{

private static final Logger logger = LoggerFactory.getLogger(TimestampInterceptor.class);

@Override
	public void close() {
	}

@Override
	public void initialize() {
	}

@Override
	public Event intercept(Event e) {
		byte[] bs=e.getBody();
		String line=null;
		if(bs!=null){
			try {
				line=new String(bs,"UTF-8");
				Mapp=e.getHeaders();
				if(p==null)
					p=new HashMap();
				String time=line.substring(0,8);
				p.put("day",time);
				e.setHeaders(p);
			} catch (Exception e1) {
				logger.error("error data:"+e+":"+line);
				return null;
			}
		}else{
			logger.error("empty data:"+e);
		}
		return e;
	}

@Override
	public List intercept(List events) {
       for (Event event : events) {  
            intercept(event);  
        }     
        return events;  
	}
	public static class Builder implements Interceptor.Builder {  
 
        @Override  
        public Interceptor build() {  
            return new TimestampInterceptor();  
        }  
  
		@Override
		public void configure(Context arg0) {
			
		}  
    }  
}
```

将上面拦截器打包成jar，放到/data/soft/flume/lib目录下即可。

在flume配置文件中指定拦截器：

```shell
agent.sources.source_xx1.interceptors=i1
agent.sources.source_xx1.interceptors.i1.type=org.apache.flume.interceptor.TimestampInterceptor$Builder
```

至此，全部完成。

注意！

本站用户原创文章，版权为本站用户，转摘请注明本文地址：http://www.silva-art.net/blog/2018/12/19/97f4ef772f92fb673a041195dd8ac6a7.html。

猜您在找

大数据应用日志采集之Scribe 安装配置指南【大数据安全】Kerberos集群安装配置大数据集群安装（一）大数据集群安装系列1：Zookeeper 的安装大数据集群搭建之节点的网络配置过程（二）