Flume + HDFS + Hive日志收集系统搭建

<!-- 产品数据分析日志 按分钟分 -->
        <RollingRandomAccessFile name="RollingFile_product_minute"
            fileName="${STAT_LOG_HOME}/${SERVER_NAME}_product.log"
            filePattern="${STAT_LOG_HOME}/${SERVER_NAME}_product.log.%d{yyyy-MM-dd-HH-mm}-%i">
            <PatternLayout charset="UTF-8"
                pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} %level - %msg%xEx%n" />
            <Policies>
                <TimeBasedTriggeringPolicy interval="1"
                    modulate="true" />
                <SizeBasedTriggeringPolicy size="${EVERY_FILE_SIZE}" />
            </Policies>
            <Filters>
                <ThresholdFilter level="INFO" onMatch="ACCEPT"
                    onMismatch="NEUTRAL" />
            </Filters>
        </RollingRandomAccessFile> 

2016-11-30 09:18:21.916 INFO - {"tableName": "ReportView",
 
    "logRequest": {***
 
    },
 
    "timestamp": 1480468701432,
 
    "statBody": {***
 
    },
 
    "logResponse": {***
 
    },
 
    "resultCode": 1,
 
    "resultFailMsg": ""
 
} 

a1.sources = linuxidcSource
 
a1.channels = linuxidcChannel
 
a1.sinks = linuxidcSink
 
a1.sources.linuxidcSource.type = spooldir
 
a1.sources.linuxidcSource.channels = linuxidcChannel
 
#日志目录
 
a1.sources.linuxidcSource.spoolDir = /opt/flumeSpool
 
a1.sources.linuxidcSource.fileHeader = true
 
#日志内容处理完后，会生成.COMPLETED 后缀的文件，同时.log 文件每一分钟 roll 一个，此处忽略.log 文件与.COMPLETED 文件
 
a1.sources.linuxidcSource.ignorePattern=([^_]+)|(.*(\.log)$)|(.*(\.COMPLETED)$)
 
a1.sources.linuxidcSource.basenameHeader=true
 
a1.sources.linuxidcSource.deserializer.maxLineLength=102400
 
#自定义拦截器，对 json 格式的源日志进行字段分隔，并添加 timestamp，为后面的 hdfsSink 做处理，拦截器代码见后面
 
a1.sources.linuxidcSource.interceptors=i1
 
a1.sources.linuxidcSource.interceptors.i1.type=com.linuxidc.flume_interceptor.HiveLogInterceptor2$Builder
 
a1.sinks.linuxidcSink.type = avro
 
a1.sinks.linuxidcSink.channel = linuxidcChannel
 
a1.sinks.linuxidcSink.hostname = centos2
 
a1.sinks.linuxidcSink.port = 4545
 
#此处配置 deflate 压缩后，hive collector 那边一定也要相应配置解压缩
 
a1.sinks.linuxidcSink.compression-type=deflate
 
a1.channels.linuxidcChannel.type=memory
 
a1.channels.linuxidcChannel.capacity=10000
 
a1.channels.linuxidcChannel.transactionCapacity=1000 

a1.sources = avroSource
 
a1.channels = memChannel
 
a1.sinks = hdfsSink
 
a1.sources.avroSource.type = avro
 
a1.sources.avroSource.channels = memChannel
 
a1.sources.avroSource.bind=centos2
 
a1.sources.avroSource.port=4545
 
#与 flume agent 配置对应
 
a1.sources.avroSource.compression-type=deflate
 
a1.sinks.hdfsSink.type = hdfs
 
a1.sinks.hdfsSink.channel = memChannel
 
# linuxidc_hive_log 为 hive 表，按年 - 月 - 日分区存储，a1.sinks.hdfsSink.hdfs.path=hdfs://centos1:9000/flume/linuxidc_hive_log/dt=%Y-%m-%d
 
a1.sinks.hdfsSink.hdfs.batchSize=10000
 
a1.sinks.hdfsSink.hdfs.fileType=DataStream
 
a1.sinks.hdfsSink.hdfs.writeFormat=Text
 
a1.sinks.hdfsSink.hdfs.rollSize=10240000
 
a1.sinks.hdfsSink.hdfs.rollCount=0
 
a1.sinks.hdfsSink.hdfs.rollInterval=300
 
a1.channels.memChannel.type=memory
 
a1.channels.memChannel.capacity=100000
 
a1.channels.memChannel.transactionCapacity=10000 

CREATE TABLE `linuxidc_hive_log`(
 
`tableNmae` string,
 
`logRequest` string,
 
`timestamp` bigint,
 
`statBody` string,
 
`logResponse` string,
 
`resultCode` int,
 
`resultFailMsg` string
 
)
 
PARTITIONED BY (`dt` string)
 
ROW FORMAT DELIMITED
 
FIELDS TERMINATED BY '\u0001'
 
STORED AS INPUTFORMAT
 
'org.apache.hadoop.mapred.TextInputFormat'
 
OUTPUTFORMAT
 
'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
 
LOCATION
 
'hdfs://centos1:9000/flume/linuxidc_hive_log'; 

for ((i=-1;i<=365;i++))
do
 
        dt=$(date -d "$(date +%F) ${i} days" +%Y-%m-%d)
 
        echo date=$dt
 
        hive -e "ALTER TABLE linuxidc_hive_log ADD PARTITION(dt='${dt}')" >> logs/init_linuxidc_hive_log.out 2>>logs/init_linuxidc_hive_log.err
 
done 

package com.linuxidc.flume_interceptor;
 
import java.util.ArrayList;
 
import java.util.List;
 
import java.util.Map;
 
import org.apache.flume.Context;
 
import org.apache.flume.Event;
 
import org.apache.flume.interceptor.Interceptor;
 
import org.apache.flume.interceptor.TimestampInterceptor.Constants;
 
import org.slf4j.Logger;
 
import org.slf4j.LoggerFactory;
 
import com.alibaba.fastjson.JSONObject;
 
import com.google.common.base.Charsets;
 
import com.google.common.base.Joiner;
 
public class HiveLogInterceptor2 implements Interceptor
 
{private static Logger logger = LoggerFactory.getLogger(HiveLogInterceptor2.class);
 
    public static final String HIVE_SEPARATOR = "\001";
 
    public void close()
 
    {// TODO Auto-generated method stub
 
    }
 
    public void initialize()
 
    {// TODO Auto-generated method stub
 
    }
 
    public Event intercept(Event event)
 
    {String orginalLog = new String(event.getBody(), Charsets.UTF_8);
 
        try
 
        {String log = this.parseLog(orginalLog);
 
            // 设置时间, 用于 hdfsSink
 
            long now = System.currentTimeMillis();
 
            Map headers = event.getHeaders();
 
            headers.put(Constants.TIMESTAMP, Long.toString(now));
 
            event.setBody(log.getBytes());
 
        } catch (Throwable throwable)
 
        {logger.error(("errror when intercept,log [" + orginalLog + "]"), throwable);
 
            return null;
 
        }
 
        return event;
 
    }
 
    public List<Event> intercept(List<Event> list)
 
    {List<Event> events = new ArrayList<Event>();
 
        for (Event event : list)
 
        {Event interceptedEvent = this.intercept(event);
 
            if (interceptedEvent != null)
 
            {events.add(interceptedEvent);
 
            }
 
        }
 
        return events;
 
    }
 
    private static String parseLog(String log)
 
    {List<String> logFileds = new ArrayList<String>();
 
        String dt = log.substring(0, 10);
 
        String keyStr = "INFO -";
 
        int index = log.indexOf(keyStr);
 
        String content = "";
 
        if (index != -1)
 
        {content = log.substring(index + keyStr.length(), log.length());
 
        }
 
        //针对不同 OS，使用不同回车换行符号
 
        content = content.replaceAll("\r", "");
 
        content = content.replaceAll("\n", "\\\\" + System.getProperty("line.separator"));
 
        JSONObject jsonObj = JSONObject.parseObject(content);
 
        String tableName = jsonObj.getString("tableName");
 
        String logRequest = jsonObj.getString("logRequest");
 
        String timestamp = jsonObj.getString("timestamp");
 
        String statBody = jsonObj.getString("statBody");
 
        String logResponse = jsonObj.getString("logResponse");
 
        String resultCode = jsonObj.getString("resultCode");
 
        String resultFailMsg = jsonObj.getString("resultFailMsg");
 
        //字段分离
 
        logFileds.add(tableName);
 
        logFileds.add(logRequest);
 
        logFileds.add(timestamp);
 
        logFileds.add(statBody);
 
        logFileds.add(logResponse);
 
        logFileds.add(resultCode);
 
        logFileds.add(resultFailMsg);
 
        logFileds.add(dt);
 
        return Joiner.on(HIVE_SEPARATOR).join(logFileds);
 
    }
 
    public static class Builder implements Interceptor.Builder
 
    {public Interceptor build()
 
        {return new HiveLogInterceptor2();}
 
        public void configure(Context arg0)
 
        {}}
 
} 

<build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-dependency-plugin</artifactId>
                <configuration>
                    <outputDirectory>
                        ${project.build.directory}
                    </outputDirectory>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-dependency-plugin</artifactId>
                <executions>
                    <execution>
                        <id>copy-dependencies</id>
                        <phase>prepare-package</phase>
                        <goals>
                            <goal>copy-dependencies</goal>
                        </goals>
                        <configuration>
                            <outputDirectory>${project.build.directory}/lib</outputDirectory>
                            <overWriteReleases>true</overWriteReleases>
                            <overWriteSnapshots>true</overWriteSnapshots>
                            <overWriteIfNewer>true</overWriteIfNewer>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build> 

ReportView^A{"request":{},"requestBody":{"detailInfos":[],"flag":"","reportId":7092,"pageSize":0,"searchs":[],"orders":[],"pageNum":1}}^A1480468701432^A{"sourceId":22745,"reportId":7092,"projectId":29355,"userId":2532}^A{"responseBody":{"statusCodeValue":200,"httpHeaders":{},"body":{"msg":" 请求成功 ","httpCode":200,"timestamp":1480468701849},"statusCode":"OK"},"response":{}}^A1^A^A2016-11-30

#!/bin/bash
jps -l|grep org.apache.flume.node.Application|awk '{print $1}'|xargs kill -9 2>&1 >/dev/null
cd "$(dirname "$0")"
cd ..
nohup bin/flume-ng agent --conf conf --conf-file conf/flume-conf.properties --name a1 2>&1 > /dev/null &

[root@centos1 bin]# rm -rf FlumeData.1480587273016.tmp 
[root@centos1 bin]# hadoop fs -ls /flume/linuxidc_hive_log/dt=2016-12-01/
Found 3 items
-rw-r--r--   3 root supergroup       5517 2016-12-01 08:12 /flume/linuxidc_hive_log/dt=2016-12-01/FlumeData.1480608753042.tmp
-rw-r--r--   3 root supergroup       5517 2016-12-01 08:40 /flume/linuxidc_hive_log/dt=2016-12-01/FlumeData.1480610453116
-rw-r--r--   3 root supergroup       5517 2016-12-01 08:44 /flume/linuxidc_hive_log/dt=2016-12-01/FlumeData.1480610453117
[root@centos1 bin]#  

[root@centos1 lib]# hive
 
Logging initialized using configuration in file:/root/apache-hive-1.2.1-bin/conf/hive-log4j.properties
hive> select * from linuxidc_hive_log limit 2;
OK
ReportView    {"request":{},"requestBody":{"detailInfos":[],"flag":"","reportId":7092,"pageSize":0,"searchs":[],"orders":[],"pageNum":1}}    1480468701432    {"sourceId":22745,"reportId":7092,"projectId":29355,"userId":2532}    {"responseBody":{"statusCodeValue":200,"httpHeaders":{},"body":{"msg":"请求成功","httpCode":200,"timestamp":1480468701849},"statusCode":"OK"},"response":{}}    1        2016-12-01
ReportDesignResult    {"request":{},"requestBody":{"sourceId":22745,"detailInfos":[{"colName":"月份 ","flag":"0","reportId":7092,"colCode":"col_2_22745","pageSize":20,"type":"1","pageNum":1,"rcolCode":"col_25538","colType":"string","formula":"","id":25538,"position":"row","colId":181664,"dorder":1,"pColName":" 月份 ","pRcolCode":"col_25538"},{"colName":" 综合利率 (合计)","flag":"1","reportId":7092,"colCode":"col_11_22745","pageSize":20,"type":"1","pageNum":1,"rcolCode":"sum_col_25539","colType":"number","formula":"sum","id":25539,"position":"group","colId":181673,"dorder":1,"pColName":" 综合利率 ","pRcolCode":"col_25539"}],"flag":"bar1","reportId":7092,"reportName":"iiiissszzzV","pageSize":100,"searchs":[],"orders":[],"pageNum":1,"projectId":29355}}    1480468703586{"reportType":"bar1","sourceId":22745,"reportId":7092,"num":5,"usedFields":" 月份 $$ 综合利率 (合计)$$","projectId":29355,"userId":2532}    {"responseBody":{"statusCodeValue":200,"httpHeaders":{},"body":{"msg":" 请求成功","reportId":7092,"httpCode":200,"timestamp":1480468703774},"statusCode":"OK"},"response":{}}    1        2016-12-01
Time taken: 2.212 seconds, Fetched: 2 row(s)
hive>

for ((i=-10;i<=365;i++))
do
 
        dt=$(date -d "$(date +%F) ${i} days" +%Y-%m-%d)
 
        echo date=$dt
 
        hive -e "ALTER TABLE linuxidc_hive_log ADD PARTITION(dt='${dt}')" >> logs/init_linuxidc_hive_log.out 2>>logs/init_linuxidc_hive_log.err
 
done