大數據離線分析平台 用戶數據Etl

Etl目標 

解析我們收集的日志數據，將解析后的數據保存到hbase中。這里選擇hbase來存儲數據的主要原因就是：

hbase的寬表結構設計適合我們的這樣多種數據格式的數據存儲（不同event有不同的存儲格式）。

在etl過程中，我們需要將我們收集得到的數據進行處理，包括ip地址解析、userAgent解析、服務器時間解析等。

 

在我們本次項目中ip解析采用的是純真ip數據庫，官網是http://www.cz88.net/
另外：ip解析可以采用淘寶提供的ip接口來進行解析
地址：http://ip.taobao.com/
接口：http://ip.taobao.com/service/getIpInfo.php?ip=[ip地址字串]

ETL存儲
etl的結果存儲到hbase中，由於考慮到不同事件有不同的數據格式，所以我們將最終etl的結果保存到hbase中，我們使用單family的數據格式，rowkey的生產模式我們采用timestamp+uuid.crc編碼的方式。hbase創建命令：create 'event_logs', 'info'

 

步驟如下：

1. 修改pom文件，添加hadoop和hbase依賴

<dependencies>
        <dependency>
            <groupId>javax.servlet</groupId>
            <artifactId>servlet-api</artifactId>
            <version>2.5</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>cz.mallat.uasparser</groupId>
            <artifactId>uasparser</artifactId>
            <version>0.6.2</version>
        </dependency>
        <!-- hadoop start -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.7.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-common</artifactId>
            <version>2.7.2</version>
        </dependency>
        <!-- hadoop end -->
        <!-- hbase start -->
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-server</artifactId>
            <version>1.1.3</version>
        </dependency>

        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-client </artifactId>
            <version>1.1.3</version>
        </dependency>


        <!-- hbase end -->
        <!-- mysql start -->
    <dependency>
      <groupId>mysql</groupId>
      <artifactId>mysql-connector-java</artifactId>
      <version>5.1.18</version>
    </dependency>
    <!-- mysql end -->    
        <dependency>
            <groupId>cz.mallat.uasparser</groupId>
            <artifactId>uasparser</artifactId>
            <version>0.6.1</version>
        </dependency>

        <dependency>
            <groupId>jdk.tools</groupId>
            <artifactId>jdk.tools</artifactId>
            <version>1.8</version>
            <scope>system</scope>
            <systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>
        </dependency>
        <dependency>
            <groupId>org.cloudera.htrace</groupId>
            <artifactId>htrace-core</artifactId>
            <version>2.04</version>
        </dependency>
        <dependency>
            <groupId>org.apache.htrace</groupId>
            <artifactId>htrace-core</artifactId>
            <version>3.1.0-incubating</version>
        </dependency>
        <dependency>
            <groupId>org.apache.htrace</groupId>
            <artifactId>htrace-core4</artifactId>
            <version>4.0.1-incubating</version>
        </dependency>
    </dependencies>

 

2. 添加LoggerUtil類,中間設計到EventLogConstant常量類和TimeUtil工具類
LoggerUtil主要作用就是解析日志，返回一個map對象

package com.kk.etl.util;

import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;

import com.kk.common.EventLogConstants;
import com.kk.etl.util.IpSeekerExt.RegionInfo;
import com.kk.etl.util.UserAgentUtil.UserAgentInfo;
import com.kk.util.TimeUtil;
public class LoggerUtils {
    private static final Logger logger = Logger.getLogger(LoggerUtils.class);
    private static IpSeekerExt ipSeekerExt = new IpSeekerExt();

    /**
     * 處理日志數據logText，返回處理結果map集合<br/>
     * 如果logText沒有指定數據格式，那么直接返回empty的集合
     * 
     * @param logText
     * @return
     */
    public static Map<String, String> handleLog(String logText) {
        Map< String, String> clientInfo=new HashMap<String,String>();
        if (logText!=null&&!logText.isEmpty()) {
            String[] splits=logText.split(EventLogConstants.LOG_SEPARTIOR);
            if (splits.length==4) {
                // 日志格式為: ip^A   服務器時間^A    host^A    請求參數
                clientInfo.put(EventLogConstants.LOG_COLUMN_NAME_IP, splits[0].trim()); // 設置ip
                //設置服務器時間
                clientInfo.put(EventLogConstants.LOG_COLUMN_NAME_SERVER_TIME, String.valueOf(TimeUtil.parseNginxServerTime2Long(splits[1].trim())));
                int index=splits[3].indexOf("?");
                if (index > -1) {
                    String requestBody = splits[3].substring(index + 1); // 獲取請求參數，也就是我們的收集數據
                    // 處理請求參數
                    handleRequestBody(requestBody, clientInfo);
                    // 處理userAgent
                    handleUserAgent(clientInfo);
                    // 處理ip地址
                    handleIp(clientInfo);
            }else {
                 // 數據格式異常
                clientInfo.clear();
            }
    
        } 
        
        
    }
        return clientInfo;
        }
        /**
         * 處理請求參數
         * 
         * @param requestBody
         * @param clientInfo
         */
        private static void handleRequestBody(String requestBody, Map<String, String> clientInfo) {
             
            String[] requestParams=requestBody.split("&");
           for (String param : requestParams) {
               if (param!=null&&!param.isEmpty()) {
                   int index=  param.indexOf("=");
                   if (index < 0) {
                       logger.warn("沒法進行解析參數:" + param + "， 請求參數為:" + requestBody);
                       continue;
                   }
                  
                try {
                     String p1=param.substring(0,index);
                      String p2 = URLDecoder.decode(param.substring(index + 1), "utf-8");
                      if (StringUtils.isNotBlank(p1) && StringUtils.isNotBlank(p2)) {
                            clientInfo.put(p1, p2);
                        }
                } catch (UnsupportedEncodingException e) {
                     logger.warn("解碼操作出現異常", e);
                     continue;
                }
                  
            } else {

            }
            
        }           
     }

    /**
     * 處理ip地址
     * 
     * @param clientInfo
     */
    private static void handleIp(Map<String,String> clientInfo) {
         if (clientInfo.containsKey(EventLogConstants.LOG_COLUMN_NAME_IP)) {
             String ip = clientInfo.get(EventLogConstants.LOG_COLUMN_NAME_IP);
             RegionInfo info = ipSeekerExt.analyticIp(ip);
             if (info!= null) {
                 if (info.getCountry().equals("unknown")&&info.getCity().equals("unknown")&&info.getProvince().equals("unknown")) {
                    info.setCountry("中國");
                    info.setProvince("廣東省");
                    info.setCity("廣州市");
                }
                 clientInfo.put(EventLogConstants.LOG_COLUMN_NAME_COUNTRY, info.getCountry());
                 clientInfo.put(EventLogConstants.LOG_COLUMN_NAME_PROVINCE, info.getProvince());
                 clientInfo.put(EventLogConstants.LOG_COLUMN_NAME_CITY, info.getCity());
             }
         }
    }

    /**
     * 處理瀏覽器的userAgent信息
     * 
     * @param clientInfo
     */
    private static void handleUserAgent(Map<String, String> clientInfo) {
        if (clientInfo.containsKey(EventLogConstants.LOG_COLUMN_NAME_USER_AGENT)) {
            UserAgentInfo info = UserAgentUtil.analyticUserAgent(clientInfo.get(EventLogConstants.LOG_COLUMN_NAME_USER_AGENT));
            if (info != null) {
                clientInfo.put(EventLogConstants.LOG_COLUMN_NAME_OS_NAME, info.getOsName());
                clientInfo.put(EventLogConstants.LOG_COLUMN_NAME_OS_VERSION, info.getOsVersion());
                clientInfo.put(EventLogConstants.LOG_COLUMN_NAME_BROWSER_NAME, info.getBrowserName());
                clientInfo.put(EventLogConstants.LOG_COLUMN_NAME_BROWSER_VERSION, info.getBrowserVersion());
            }
        }
    }

    
    

}

 

EventLogConstants主要作用就是描述hbase的event_logs表的信息(表名，列簇名，列名)以及日志收集的日志中的數據參數name。其實列名和name是一樣的。

package com.kk.common;

import java.nio.ByteBuffer;

/**
 * 
 * @author hzk
 *@parm
 *
 */
public class EventLogConstants {
    
    /**
     * 事件枚舉類。指定事件的名稱
     * 
     * @author gerry
     *
     */
    public static enum EventEnum {
        LAUNCH(1, "launch event", "e_l"), // launch事件，表示第一次訪問
        PAGEVIEW(2, "page view event", "e_pv"), // 頁面瀏覽事件
        CHARGEREQUEST(3, "charge request event", "e_crt"), // 訂單生產事件
        CHARGESUCCESS(4, "charge success event", "e_cs"), // 訂單成功支付事件
        CHARGEREFUND(5, "charge refund event", "e_cr"), // 訂單退款事件
        EVENT(6, "event duration event", "e_e") // 事件
        ;

        public final int id; // id 唯一標識
        public final String name; // 名稱
        public final String alias; // 別名，用於數據收集的簡寫

        private EventEnum(int id, String name, String alias) {
            this.id = id;
            this.name = name;
            this.alias = alias;
        }

        /**
         * 獲取匹配別名的event枚舉對象，如果最終還是沒有匹配的值，那么直接返回null。
         * 
         * @param alias
         * @return
         */
        public static EventEnum valueOfAlias(String alias) {
            for (EventEnum event : values()) {
                if (event.alias.equals(alias)) {
                    return event;
                }
            }
            return null;
        }
    }

    /**
     * 表名稱
     */
    public static final String HBASE_NAME_EVENT_LOGS = "event_logs";

    /**
     * event_logs表的列簇名稱
     */
    public static final String EVENT_LOGS_FAMILY_NAME = "info";

    /**
     * 日志分隔符
     */
    public static final String LOG_SEPARTIOR = "\\^A";

    /**
     * 用戶ip地址
     */
    public static final String LOG_COLUMN_NAME_IP = "ip";

    /**
     * 服務器時間
     */
    public static final String LOG_COLUMN_NAME_SERVER_TIME = "s_time";

    /**
     * 事件名稱
     */
    public static final String LOG_COLUMN_NAME_EVENT_NAME = "en";

    /**
     * 數據收集端的版本信息
     */
    public static final String LOG_COLUMN_NAME_VERSION = "ver";

    /**
     * 用戶唯一標識符
     */
    public static final String LOG_COLUMN_NAME_UUID = "u_ud";

    /**
     * 會員唯一標識符
     */
    public static final String LOG_COLUMN_NAME_MEMBER_ID = "u_mid";

    /**
     * 會話id
     */
    public static final String LOG_COLUMN_NAME_SESSION_ID = "u_sd";
    /**
     * 客戶端時間
     */
    public static final String LOG_COLUMN_NAME_CLIENT_TIME = "c_time";
    /**
     * 語言
     */
    public static final String LOG_COLUMN_NAME_LANGUAGE = "l";
    /**
     * 瀏覽器user agent參數
     */
    public static final String LOG_COLUMN_NAME_USER_AGENT = "b_iev";
    /**
     * 瀏覽器分辨率大小
     */
    public static final String LOG_COLUMN_NAME_RESOLUTION = "b_rst";
    /**
     * 當前url
     */
    public static final String LOG_COLUMN_NAME_CURRENT_URL = "p_url";
    /**
     * 前一個頁面的url
     */
    public static final String LOG_COLUMN_NAME_REFERRER_URL = "p_ref";
    /**
     * 當前頁面的title
     */
    public static final String LOG_COLUMN_NAME_TITLE = "tt";
    /**
     * 訂單id
     */
    public static final String LOG_COLUMN_NAME_ORDER_ID = "oid";
    /**
     * 訂單名稱
     */
    public static final String LOG_COLUMN_NAME_ORDER_NAME = "on";
    /**
     * 訂單金額
     */
    public static final String LOG_COLUMN_NAME_ORDER_CURRENCY_AMOUNT = "cua";
    /**
     * 訂單貨幣類型
     */
    public static final String LOG_COLUMN_NAME_ORDER_CURRENCY_TYPE = "cut";
    /**
     * 訂單支付金額
     */
    public static final String LOG_COLUMN_NAME_ORDER_PAYMENT_TYPE = "pt";
    /**
     * category名稱
     */
    public static final String LOG_COLUMN_NAME_EVENT_CATEGORY = "ca";
    /**
     * action名稱
     */
    public static final String LOG_COLUMN_NAME_EVENT_ACTION = "ac";
    /**
     * kv前綴
     */
    public static final String LOG_COLUMN_NAME_EVENT_KV_START = "kv_";
    /**
     * duration持續時間
     */
    public static final String LOG_COLUMN_NAME_EVENT_DURATION = "du";
    /**
     * 操作系統名稱
     */
    public static final String LOG_COLUMN_NAME_OS_NAME = "os";
    /**
     * 操作系統版本
     */
    public static final String LOG_COLUMN_NAME_OS_VERSION = "os_v";
    /**
     * 瀏覽器名稱
     */
    public static final String LOG_COLUMN_NAME_BROWSER_NAME = "browser";
    /**
     * 瀏覽器版本
     */
    public static final String LOG_COLUMN_NAME_BROWSER_VERSION = "browser_v";
    /**
     * ip地址解析的所屬國家
     */
    public static final String LOG_COLUMN_NAME_COUNTRY = "country";
    /**
     * ip地址解析的所屬省份
     */
    public static final String LOG_COLUMN_NAME_PROVINCE = "province";
    /**
     * ip地址解析的所屬城市
     */
    public static final String LOG_COLUMN_NAME_CITY = "city";

    /**
     * 定義platform
     */
    public static final String LOG_COLUMN_NAME_PLATFORM = "pl";

}

 

TimeUtil主要作用解析服務器時間以及定義rowkey中的timestamp時間戳格式。

package com.kk.util;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.lang.StringUtils;

import com.kk.common.DateEnum;

public class TimeUtil {

    /**
     * 將nginx服務器時間轉換為時間戳，如果說解析失敗，返回-1
     * 
     * @param input
     * @return
     */
    public static long parseNginxServerTime2Long(String input) {
        Date date = parseNginxServerTime2Date(input);
        return date == null ? -1L : date.getTime();
    }

    /**
     * 將nginx服務器時間轉換為date對象。如果解析失敗，返回null
     * 
     * @param input
     *            格式: 1449410796.976
     * @return
     */
    public static Date parseNginxServerTime2Date(String input) {
        if (StringUtils.isNotBlank(input)) {
            try {
                long timestamp = Double.valueOf(Double.valueOf(input.trim()) * 1000).longValue();
                Calendar calendar = Calendar.getInstance();
                calendar.setTimeInMillis(timestamp);
                return calendar.getTime();
            } catch (Exception e) {
                // nothing
            }
        }
        return null;
    }

    /**
     * 判斷輸入的參數是否是一個有效的時間格式數據
     * 
     * @param input
     * @return
     */
    public static boolean isValidateRunningDate(String input) {
        Matcher matcher = null;
        boolean result = false;
        String regex = "[0-9]{4}-[0-9]{2}-[0-9]{2}";
        if (input != null && !input.isEmpty()) {
            Pattern pattern = Pattern.compile(regex);
            matcher = pattern.matcher(input);
        }
        if (matcher != null) {
            result = matcher.matches();
        }
        return result;
    }
    public static final String DATE_FORMAT = "yyyy-MM-dd";

    /**
     * 獲取昨日的日期格式字符串數據
     * 
     * @return
     */
    public static String getYesterday() {
        return getYesterday(DATE_FORMAT);
    }

    /**
     * 獲取對應格式的時間字符串
     * 
     * @param pattern
     * @return
     */
    public static String getYesterday(String pattern) {
        SimpleDateFormat sdf = new SimpleDateFormat(pattern);
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.DAY_OF_YEAR, -1);
        return sdf.format(calendar.getTime());
    }

    /**
     * 將yyyy-MM-dd格式的時間字符串轉換為時間戳
     * 
     * @param input
     * @return
     */
    public static long parseString2Long(String input) {
        return parseString2Long(input, DATE_FORMAT);
    }
    /**
     * 將指定格式的時間字符串轉換為時間戳
     * 
     * @param input
     * @param pattern
     * @return
     */
    public static long parseString2Long(String input, String pattern) {
        Date date = null;
        try {
            date = new SimpleDateFormat(pattern).parse(input);
        } catch (ParseException e) {
            throw new RuntimeException(e);
        }
        return date.getTime();
    }

    /**
     * 將時間戳轉換為yyyy-MM-dd格式的時間字符串
     * @param input
     * @return
     */
    public static String parseLong2String(long input) {
        return parseLong2String(input, DATE_FORMAT);
    }

    /**
     * 將時間戳轉換為指定格式的字符串
     * 
     * @param input
     * @param pattern
     * @return
     */
    public static String parseLong2String(long input, String pattern) {
        Calendar calendar = Calendar.getInstance();
        calendar.setTimeInMillis(input);
        return new SimpleDateFormat(pattern).format(calendar.getTime());
    }

    /**
     * 從時間戳中獲取需要的時間信息
     * 
     * @param time
     *            時間戳
     * @param type
     * @return 如果沒有匹配的type，拋出異常信息
     */
    public static int getDateInfo(long time, DateEnum type) {
        Calendar calendar = Calendar.getInstance();
        calendar.setTimeInMillis(time);
        if (DateEnum.YEAR.equals(type)) {
            // 需要年份信息
            return calendar.get(Calendar.YEAR);
        } else if (DateEnum.SEASON.equals(type)) {
            // 需要季度信息
            int month = calendar.get(Calendar.MONTH) + 1;
            if (month % 3 == 0) {
                return month / 3;
            }
            return month / 3 + 1;
        } else if (DateEnum.MONTH.equals(type)) {
            // 需要月份信息
            return calendar.get(Calendar.MONTH) + 1;
        } else if (DateEnum.WEEK.equals(type)) {
            // 需要周信息
            return calendar.get(Calendar.WEEK_OF_YEAR);
        } else if (DateEnum.DAY.equals(type)) {
            return calendar.get(Calendar.DAY_OF_MONTH);
        } else if (DateEnum.HOUR.equals(type)) {
            return calendar.get(Calendar.HOUR_OF_DAY);
        }
        throw new RuntimeException("沒有對應的時間類型:" + type);
    }

    /**
     * 獲取time指定周的第一天的時間戳值
     * 
     * @param time
     * @return
     */
    public static long getFirstDayOfThisWeek(long time) {
        Calendar cal = Calendar.getInstance();
        cal.setTimeInMillis(time);
        cal.set(Calendar.DAY_OF_WEEK, 1);
        cal.set(Calendar.HOUR_OF_DAY, 0);
        cal.set(Calendar.MINUTE, 0);
        cal.set(Calendar.SECOND, 0);
        cal.set(Calendar.MILLISECOND, 0);
        return cal.getTimeInMillis();
    }
}

 

3. 編寫mapper類和runner類

mapper

package com.kk.etl.util.mr.ald;

import java.io.IOException;
import java.util.Map;
import java.util.zip.CRC32;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.log4j.Logger;

import com.kk.common.EventLogConstants;
import com.kk.common.EventLogConstants.EventEnum;
import com.kk.etl.util.LoggerUtils;

/**
 * 自定義數據解析map類
 * 
 * @author gerry
 *
 */
public class AnalyserLogDataMapper extends Mapper<Object, Text, NullWritable, Put> {
    
     private final Logger logger = Logger.getLogger(AnalyserLogDataMapper.class);
        private int inputRecords, filterRecords, outputRecords; // 主要用於標志，方便查看過濾數據
        private byte[] family = Bytes.toBytes(EventLogConstants.EVENT_LOGS_FAMILY_NAME);
        private CRC32 crc32 = new CRC32();

        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            this.inputRecords++;
            this.logger.debug("Analyse data of :" + value);

            try {
                // 解析日志
                Map<String, String> clientInfo = LoggerUtils.handleLog(value.toString());

                // 過濾解析失敗的數據
                if (clientInfo.isEmpty()) {
                    this.filterRecords++;
                    return;
                }

                // 獲取事件名稱
                String eventAliasName = clientInfo.get(EventLogConstants.LOG_COLUMN_NAME_EVENT_NAME);
                EventEnum event = EventEnum.valueOfAlias(eventAliasName);
                switch (event) {
                case LAUNCH:
                case PAGEVIEW:
                case CHARGEREQUEST:
                case CHARGEREFUND:
                case CHARGESUCCESS:
                case EVENT:
                    // 處理數據
                    this.handleData(clientInfo, event, context);
                    break;
                default:
                    this.filterRecords++;
                    this.logger.warn("該事件沒法進行解析，事件名稱為:" + eventAliasName);
                }
            } catch (Exception e) {
                this.filterRecords++;
                this.logger.error("處理數據發出異常，數據:" + value, e);
            }
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            super.cleanup(context);
            logger.info("輸入數據:" + this.inputRecords + "；輸出數據:" + this.outputRecords + "；過濾數據:" + this.filterRecords);
        }

        /**
         * 具體處理數據的方法
         * 
         * @param clientInfo
         * @param context
         * @param event
         * @throws InterruptedException
         * @throws IOException
         */
        private void handleData(Map<String, String> clientInfo, EventEnum event, Context context) throws IOException, InterruptedException {
            String uuid = clientInfo.get(EventLogConstants.LOG_COLUMN_NAME_UUID);
            String memberId = clientInfo.get(EventLogConstants.LOG_COLUMN_NAME_MEMBER_ID);
            String serverTime = clientInfo.get(EventLogConstants.LOG_COLUMN_NAME_SERVER_TIME);
            if (StringUtils.isNotBlank(serverTime)) {
                // 要求服務器時間不為空
                clientInfo.remove(EventLogConstants.LOG_COLUMN_NAME_USER_AGENT); // 瀏覽器信息去掉
                String rowkey = this.generateRowKey(uuid, memberId, event.alias, serverTime); // timestamp
                                                                                              // +
                                                                                              // (uuid+memberid+event).crc
                Put put = new Put(Bytes.toBytes(rowkey));
                for (Map.Entry<String, String> entry : clientInfo.entrySet()) {
                    if (StringUtils.isNotBlank(entry.getKey()) && StringUtils.isNotBlank(entry.getValue())) {
                        put.add(family, Bytes.toBytes(entry.getKey()), Bytes.toBytes(entry.getValue()));
                    }
                }
                context.write(NullWritable.get(), put);
                this.outputRecords++;
            } else {
                this.filterRecords++;
            }
        }

        /**
         * 根據uuid memberid servertime創建rowkey
         * 
         * @param uuid
         * @param memberId
         * @param eventAliasName
         * @param serverTime
         * @return
         */
        private String generateRowKey(String uuid, String memberId, String eventAliasName, String serverTime) {
            StringBuilder sb = new StringBuilder();
            sb.append(serverTime).append("_");
            this.crc32.reset();
            if (StringUtils.isNotBlank(uuid)) {
                this.crc32.update(uuid.getBytes());
            }
            if (StringUtils.isNotBlank(memberId)) {
                this.crc32.update(memberId.getBytes());
            }
            this.crc32.update(eventAliasName.getBytes());
            sb.append(this.crc32.getValue() % 100000000L);
            return sb.toString();
        }
    
    
    
}

 

runner

package com.kk.etl.util.mr.ald;
import java.io.File;
import java.io.IOException;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Logger;

import com.kk.common.EventLogConstants;
import com.kk.common.GlobalConstants;
import com.kk.util.EJob;
import com.kk.util.TimeUtil;

import cz.mallat.uasparser.*;

public class AnalyserLogDataRunner implements Tool{
    private static final Logger logger = Logger.getLogger(AnalyserLogDataRunner.class);
    private Configuration conf = null;

    public static void main(String[] args) {
        try {
            ToolRunner.run(new Configuration(), new AnalyserLogDataRunner(), args);
        } catch (Exception e) {
            logger.error("執行日志解析job異常", e);
            throw new RuntimeException(e);
        }
    }

    @Override
    public void setConf(Configuration conf) {
        this.conf = HBaseConfiguration.create(conf);
    }

    @Override
    public Configuration getConf() {
        return this.conf;
    }

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = this.getConf();
        this.processArgs(conf, args);
        conf.set("fs.defaultFS","hdfs://hadoop-001:9000");
        conf.set("mapreduce.framework.name","yarn");
        conf.set("yarn.resourcemanager.hostname","hadoop-002");

        Job job = Job.getInstance(conf, "analyser_logdata" );

        // 設置本地提交job，集群運行，需要代碼
        File jarFile = EJob.createTempJar("target/classes");
        JobConf jobConf=(JobConf)job.getConfiguration();
        jobConf.setJar(jarFile.toString());
        // 設置本地提交job，集群運行，需要代碼結束

        job.setJarByClass(AnalyserLogDataRunner.class);
        job.setMapperClass(AnalyserLogDataMapper.class);
        job.setMapOutputKeyClass(NullWritable.class);
        job.setMapOutputValueClass(Put.class);
       
        // 設置reducer配置
//         1. 集群上運行，打成jar運行(要求addDependencyJars參數為true，默認就是true)
        conf.set("addDependencyJars", "true");
        TableMapReduceUtil.initTableReducerJob(EventLogConstants.HBASE_NAME_EVENT_LOGS, null, job);
        // 2. 本地運行，要求參數addDependencyJars為false
//     
//        conf.set("addDependencyJars", "false");
//        TableMapReduceUtil.initTableReducerJob(EventLogConstants.HBASE_NAME_EVENT_LOGS,
//         null, job, null, null, null, null, false);
        job.setNumReduceTasks(0);

        // 設置輸入路徑
        this.setJobInputPaths(job);
        return job.waitForCompletion(true) ? 0 : -1;
    }

    /**
     * 處理參數
     * 
     * @param conf
     * @param args
     */
    private void processArgs(Configuration conf, String[] args) {
        String date = null;
        for (int i = 0; i < args.length; i++) {
            if ("-d".equals(args[i])) {
                if (i + 1 < args.length) {
                    date = args[++i];
                    break;
                }
            }
        }

        // 要求date格式為: yyyy-MM-dd
        if (StringUtils.isBlank(date) || !TimeUtil.isValidateRunningDate(date)) {
            // date是一個無效時間數據
            date = TimeUtil.getYesterday(); // 默認時間是昨天
        }
        conf.set(GlobalConstants.RUNNING_DATE_PARAMES, date);
    }

    /**
     * 設置job的輸入路徑
     * 
     * @param job
     */
    private void setJobInputPaths(Job job) {
        Configuration conf = job.getConfiguration();
        FileSystem fs = null;
        try {
            fs = FileSystem.get(conf);
            String date = conf.get(GlobalConstants.RUNNING_DATE_PARAMES);
            Path inputPath = new Path("/logs/nginx/" + TimeUtil.parseLong2String(TimeUtil.parseString2Long(date), "MM/dd"));//+"/BF-01.1553792641215");
            if (fs.exists(inputPath)) {
                FileInputFormat.addInputPath(job, inputPath);
            } else {
                throw new RuntimeException("文件不存在:" + inputPath);
            }
        } catch (IOException e) {
            throw new RuntimeException("設置job的mapreduce輸入路徑出現異常", e);
        } finally {
            if (fs != null) {
                try {
                    fs.close();
                } catch (IOException e) {
                    // nothing
                }
            }
        }
    }

}

 

4. 添加環境變量文件，core-site.xml hbase-site.xml log4j.properties 根據不同的運行情況，修改源碼將修改后的源碼放到代碼中。

5. 添加pom編譯代碼，並進行測試
本地運行測試： 需要注意的就是windows環境可能會導致出現access方法調用異常，需要修改nativeio這個java文件。
使用TableMapReduceUtil的時候如果出現異常：

 

---

*****/htrace-core-2.04.jar from hdfs://***/htrace-core-2.04.jar is not a valid DFS filename.

---

 

就需要將addDependencyJars參數設置為false。
本地提交job，集群運行測試：
本地需要知道提交的job是需要提交到集群上的，所以需要指定兩個參數mapreduce.framework.name和yarn.resourcemanager.address，value分別為yarn和hh:8032即可，但是可能會出現異常信息，此時需要將參數mapreduce.app-submission.cross-platform設置為true。

 

參數設置：

---

mapreduce.framework.name=yarn
yarn.resourcemanager.address=hadoop-001:8032
mapreduce.app-submission.cross-platform=true

---

目錄結構如下

 

 

異常：
1. Permission denied: user=gerry, access=EXECUTE, inode="/tmp":hadoop:supergroup:drwx------
解決方案：執行

 

---

hdfs dfs -chmod -R 777 /tmp

---

2. Stack trace: ExitCodeException exitCode=1: /bin/bash: line 0: fg: no job control
解決方案：

---

添加mapreduce.app-submission.cross-platform=true

---

3. ExitCodeException exitCode=1:
解決方案：

---

habse指定輸出reducer的時候必須給定addDependencyJars參數為true。

---

4. Class com.beifeng.etl.mr.ald.AnalyserLogDataMapper not found
解決方案：

 

---

引入EJob.java文件，然后再runner類中添加代碼
File jarFile = EJob.createTempJar("target/classes");
((JobConf) job.getConfiguration()).setJar(jarFile.toString());

---

集群提交&運行job測試：