天天看點
傳回

項目實戰從0到1之大資料項目之電商數倉（使用者行為資料采集四）

作者：極目館主

4.4 采集日志Flume

項目實戰從0到1之大資料項目之電商數倉（使用者行為資料采集四）

4.4.1 日志采集Flume安裝

叢集規劃：

項目實戰從0到1之大資料項目之電商數倉（使用者行為資料采集四）

4.4.2 項目經驗之Flume元件

1）Source （1）Taildir Source相比Exec Source、Spooling Directory Source的優勢 TailDir Source：斷點續傳、多目錄。Flume1.6以前需要自己自定義Source記錄每次讀取檔案位置，實作斷點續傳。 Exec Source可以實時搜集資料，但是在Flume不運作或者Shell指令出錯的情況下，資料将會丢失。 Spooling Directory Source監控目錄，不支援斷點續傳。 （2）batchSize大小如何設定？ 答：Event 1K左右時，500-1000合适（預設為100） 2）Channel 采用Kafka Channel，省去了Sink，提高了效率。

4.4.3 日志采集Flume配置

1）Flume配置分析

項目實戰從0到1之大資料項目之電商數倉（使用者行為資料采集四）

Flume直接讀log日志的資料，log日志的格式是app-yyyy-mm-dd.log。 2）Flume的具體配置如下： （1）在/opt/module/flume/conf目錄下建立file-flume-kafka.conf檔案

[kgg@hadoop101 conf]$ vim file-flume-kafka.conf
在檔案配置如下内容
在檔案配置如下内容
a1.sources=r1
a1.channels=c1 c2

# configure source
a1.sources.r1.type = TAILDIR
a1.sources.r1.positionFile = /opt/module/flume/test/log_position.json
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 = /tmp/logs/app.+
a1.sources.r1.fileHeader = true
a1.sources.r1.channels = c1 c2
​
#interceptor
a1.sources.r1.interceptors =  i1 i2
a1.sources.r1.interceptors.i1.type = com.kgg.flume.interceptor.LogETLInterceptor$Builder
a1.sources.r1.interceptors.i2.type = com.kgg.flume.interceptor.LogTypeInterceptor$Builder
​
a1.sources.r1.selector.type = multiplexing
a1.sources.r1.selector.header = topic
a1.sources.r1.selector.mapping.topic_start = c1
a1.sources.r1.selector.mapping.topic_event = c2
​
# configure channel
a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.c1.kafka.bootstrap.servers = hadoop101:9092,hadoop102:9092,hadoop103:9092
a1.channels.c1.kafka.topic = topic_start
a1.channels.c1.parseAsFlumeEvent = false
a1.channels.c1.kafka.consumer.group.id = flume-consumer
​
a1.channels.c2.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.c2.kafka.bootstrap.servers = hadoop101:9092,hadoop102:9092,hadoop103:9092
a1.channels.c2.kafka.topic = topic_event
a1.channels.c2.parseAsFlumeEvent = false
a1.channels.c2.kafka.consumer.group.id = flume-consumer

注意：com.kgg.flume.interceptor.LogETLInterceptor和com.kgg.flume.interceptor.LogTypeInterceptor是自定義的攔截器的全類名。需要根據使用者自定義的攔截器做相應修改。

項目實戰從0到1之大資料項目之電商數倉（使用者行為資料采集四）

4.4.4 Flume的ETL和分類型攔截器

本項目中自定義了兩個攔截器，分别是：ETL攔截器、日志類型區分攔截器。 ETL攔截器主要用于，過濾時間戳不合法和Json資料不完整的日志

日志類型區分攔截器主要用于，将啟動日志和事件日志區分開來，友善發往Kafka的不同Topic。

1）建立Maven工程flume-interceptor

2）建立包名：com.kgg.flume.interceptor

3）在pom.xml檔案中添加如下配置

<dependencies>
    <dependency>
        <groupId>org.apache.flume</groupId>
        <artifactId>flume-ng-core</artifactId>
        <version>1.7.0</version>
    </dependency>
</dependencies>
​
<build>
    <plugins>
        <plugin>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>2.3.2</version>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
            </configuration>
        </plugin>
        <plugin>
            <artifactId>maven-assembly-plugin</artifactId>
            <configuration>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
            </configuration>
            <executions>
                <execution>
                    <id>make-assembly</id>
                    <phase>package</phase>
                    <goals>
                        <goal>single</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

4）在com.kgg.flume.interceptor包下建立LogETLInterceptor類名

Flume ETL攔截器LogETLInterceptor
package com.kgg.flume.interceptor;
​
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
​
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
​
public class LogETLInterceptor implements Interceptor {
​
    @Override
    public void initialize() {
​
    }
​
    @Override
    public Event intercept(Event event) {
​
        // 1 擷取資料
        byte[] body = event.getBody();
        String log = new String(body, Charset.forName("UTF-8"));
​
        // 2 判斷資料類型并向Header中指派
        if (log.contains("start")) {
            if (LogUtils.validateStart(log)){
                return event;
            }
        }else {
            if (LogUtils.validateEvent(log)){
                return event;
            }
        }
​
        // 3 傳回校驗結果
        return null;
    }
​
    @Override
    public List<Event> intercept(List<Event> events) {
​
        ArrayList<Event> interceptors = new ArrayList<>();
​
        for (Event event : events) {
            Event intercept1 = intercept(event);
​
            if (intercept1 != null){
                interceptors.add(intercept1);
            }
        }
​
        return interceptors;
    }
​
    @Override
    public void close() {
​
    }
​
    public static class Builder implements Interceptor.Builder{
​
        @Override
        public Interceptor build() {
            return new LogETLInterceptor();
        }
​
        @Override
        public void configure(Context context) {
​
        }
    }
}

4）Flume日志過濾工具類

package com.kgg.flume.interceptor;
import org.apache.commons.lang.math.NumberUtils;
​
public class LogUtils {
​
    public static boolean validateEvent(String log) {
        // 伺服器時間 | json
        // 1549696569054 | {"cm":{"ln":"-89.2","sv":"V2.0.4","os":"8.2.0","g":"[email protected]","nw":"4G","l":"en","vc":"18","hw":"1080*1920","ar":"MX","uid":"u8678","t":"1549679122062","la":"-27.4","md":"sumsung-12","vn":"1.1.3","ba":"Sumsung","sr":"Y"},"ap":"weather","et":[]}
​
        // 1 切割
        String[] logContents = log.split("\\|");
​
        // 2 校驗
        if(logContents.length != 2){
            return false;
        }
​
        //3 校驗伺服器時間
        if (logContents[0].length()!=13 || !NumberUtils.isDigits(logContents[0])){
            return false;
        }
​
        // 4 校驗json
        if (!logContents[1].trim().startsWith("{") || !logContents[1].trim().endsWith("}")){
            return false;
        }
​
        return true;
    }
​
    public static boolean validateStart(String log) {
 // {"action":"1","ar":"MX","ba":"HTC","detail":"542","en":"start","entry":"2","extend1":"","g":"[email protected]","hw":"640*960","l":"en","la":"-43.4","ln":"-98.3","loading_time":"10","md":"HTC-5","mid":"993","nw":"WIFI","open_ad_type":"1","os":"8.2.1","sr":"D","sv":"V2.9.0","t":"1559551922019","uid":"993","vc":"0","vn":"1.1.5"}
​
        if (log == null){
            return false;
        }
​
        // 校驗json
        if (!log.trim().startsWith("{") || !log.trim().endsWith("}")){
            return false;
        }
​
        return true;
    }
}

5）Flume日志類型區分攔截器LogTypeInterceptor

package com.kgg.flume.interceptor;
​
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
​
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
​
public class LogTypeInterceptor implements Interceptor {
    @Override
    public void initialize() {
​
    }
​
    @Override
    public Event intercept(Event event) {
​
        // 區分日志類型：   body  header
        // 1 擷取body資料
        byte[] body = event.getBody();
        String log = new String(body, Charset.forName("UTF-8"));
​
        // 2 擷取header
        Map<String, String> headers = event.getHeaders();
​
        // 3 判斷資料類型并向Header中指派
        if (log.contains("start")) {
            headers.put("topic","topic_start");
        }else {
            headers.put("topic","topic_event");
        }
​
        return event;
    }
​
    @Override
    public List<Event> intercept(List<Event> events) {
​
        ArrayList<Event> interceptors = new ArrayList<>();
​
        for (Event event : events) {
            Event intercept1 = intercept(event);
​
            interceptors.add(intercept1);
        }
​
        return interceptors;
    }
​
    @Override
    public void close() {
​
    }
​
    public static class Builder implements  Interceptor.Builder{
​
        @Override
        public Interceptor build() {
            return new LogTypeInterceptor();
        }
​
        @Override
        public void configure(Context context) {
​
        }
    }
}

6）打包 攔截器打包之後，隻需要單獨包，不需要将依賴的包上傳。打包之後要放入Flume的lib檔案夾下面。

項目實戰從0到1之大資料項目之電商數倉（使用者行為資料采集四）

注意：為什麼不需要依賴包？因為依賴包在flume的lib目錄下面已經存在了。

7）需要先将打好的包放入到hadoop101的/opt/module/flume/lib檔案夾下面。

ls | grep interceptor
flume-interceptor-1.0-SNAPSHOT.jar

4.4.5 日志采集Flume啟動停止腳本

1）在/home/kgg/bin目錄下建立腳本f1.sh

vim f1.sh    
在腳本中填寫如下内容
#! /bin/bash
​
case $1 in
"start"){
        for i in hadoop101 hadoop102
        do
                echo " --------啟動 $i 采集flume-------"
                ssh $i "nohup /opt/module/flume/bin/flume-ng agent --conf-file /opt/module/flume/conf/file-flume-kafka.conf --name a1 -Dflume.root.logger=INFO,LOGFILE > /dev/null 2>&1 &"
        done
};;    
"stop"){
        for i in hadoop101 hadoop102
        do
                echo " --------停止 $i 采集flume-------"
                ssh $i "ps -ef | grep file-flume-kafka | grep -v grep |awk '{print \$2}' | xargs kill"
        done
​
};;
esac

說明1：nohup，該指令可以在你退出帳戶/關閉終端之後繼續運作相應的程序。nohup就是不挂起的意思，不挂斷地運作指令。 說明2：/dev/null代表linux的空裝置檔案，所有往這個檔案裡面寫入的内容都會丢失，俗稱“黑洞”。 标準輸入0：從鍵盤獲得輸入 /proc/self/fd/0 标準輸出1：輸出到螢幕（即控制台） /proc/self/fd/1 錯誤輸出2：輸出到螢幕（即控制台） /proc/self/fd/2 2）增加腳本執行權限

chmod 777 f1.sh

3）f1叢集啟動腳本

f1.sh start

4）f1叢集停止腳本

f1.sh stop
使用者行為資料 大資料項目 實戰從0到1 項目實戰 使用者行為 資料項目 資料采集 從0到1
上一篇: java的輸入輸出函數_Java輸入輸出流的使用詳細介紹
下一篇: java中常見的輸入輸出-Scanner類

繼續閱讀