大數據技術之_25_手機APP信息統計系統項目_01_APP 數據生成模塊 + 數據收集模塊 + 數據處理模塊框架搭建 + 業務需求處理 + 數據展現模塊 +項目總結 + 問題總結

時間 2019-12-05

標籤數據技術手機 app 信息統計系統項目生成模塊收集數據處理框架搭建業務需求處理展現總結問題欄目大數據简体版

原文原文鏈接

一項目概述

1.1 角色

1）App 開發商
每一個開發商能夠有多個 App 產品。
2）App 軟件
3）數據服務平臺提供商
友盟：向 App 開發商提供服務。提供 App 使用狀況的統計服務。
友盟官方網址：https://www.umeng.com/
4）SDK
數據服務平臺提供商提供給 App 開發商的軟件包。
內置 log 上報程序。
5）用戶
每一個使用 App 的設備。
6）租戶
購買了數據服務平臺提供商服務的 App 開發商。php

1.2 業務術語

1）用戶
用戶以設備爲判斷標準，在移動統計中，每一個獨立設備認爲是一個獨立用戶。Android 系統根據 IMEI 號，IOS 系統根據 OpenUDID 來標識一個獨立用戶，每部手機一個用戶。
2）新增用戶
首次聯網使用應用的用戶。若是一個用戶首次打開某 app，那這個用戶定義爲新增用戶；卸載再安裝的設備，不會被算做一次新增。日新增用戶、周新增用戶、月新增用戶。
3）活躍用戶
打開應用的用戶即爲活躍用戶，不考慮用戶的使用狀況。天天一臺設備打開屢次會被計爲一個活躍用戶。
4）周（月）活躍用戶
某個天然周（月）內啓動過應用的用戶，該周（月）內的屢次啓動只記一個活躍用戶。
5）月活躍率
月活躍用戶與截止到該月累計的用戶總和之間的比例。
6）沉默用戶
用戶僅在安裝當天（第二天）啓動一次，後續時間無再啓動行爲。該指標能夠反映新增用戶質量和用戶與 APP 的匹配程度。
7）版本分佈
不一樣版本的周內各天新增用戶數，活躍用戶數和啓動次數。利於判斷 App 各個版本之間的優劣和用戶行爲習慣。
8）本週迴流用戶
上週未啓動過應用，本週啓動了應用的用戶。
9）連續 n 周活躍用戶
連續 n 周，每週至少啓動一次。
10）忠誠用戶
連續活躍 5 周以上的用戶
11）連續活躍用戶
連續 2 周及以上活躍的用戶
12）近期流失用戶
連續 n(2 <= n <= 4) 周沒有啓動應用的用戶。（第 n+1 周沒有啓動過）
13）留存用戶
某段時間內的新增用戶，通過一段時間後，仍然使用應用的被認做是留存用戶；這部分用戶佔當時新增用戶的比例便是留存率。例如，5 月份新增用戶 200，這 200 人在 6 月份啓動過應用的有 100 人，7 月份啓動過應用的有 80 人，8 月份啓動過應用的有 50 人；則 5 月份新增用戶一個月後的留存率是 50%，二個月後的留存率是 40%，三個月後的留存率是 25%。
14）用戶新鮮度
天天啓動應用的新老用戶比例。
15）單次使用時長
每次啓動使用的時間長度。
16）日使用時長
累計一天內的使用時間長度。
17）啓動次數計算標準
IOS 平臺應用退到後臺就算一次獨立的啓動；Android 平臺咱們規定，兩次啓動之間的間隔小於 30 秒，被計算一次啓動。用戶在使用過程當中，若因收發短信或接電話等退出應用 30 秒又再次返回應用中，那這兩次行爲應該是延續而非獨立的，因此能夠被算做一次使用行爲，即一次啓動。業內大多使用 30 秒這個標準，但用戶仍是能夠自定義此時間間隔。css

1.3 項目效果展現

一、日活用戶統計html

二、新增用戶統計

二項目需求

1）實現收集手機 APP 日誌。
2）按期離線分析手機 APP 新增用戶、活躍用戶、沉默用戶、啓動次數、版本分佈、和留存用戶等業務指標。
3）在數據展現服務上能夠查詢結果。java

三項目概要

3.1 項目技術架構

1）手機 APP 啓動時，上報啓動日誌、錯誤日誌、頁面日誌、事件日誌、使用時長日誌等信息到日誌收集服務器。
2）日誌收集服務器將收集到的日誌信息發送給 kafka。
3）Flume 分別消費 kafka 中的 5 種主題信息，並把數據存儲到 HDFS 上。
4）經過 crontab 任務調度定時把 HDFS 中的信息拷貝到 Hive 數據倉庫中。
5）核心業務操做採用 Hive 查詢。
6）查詢結果經過數據展現平臺展現。

3.2 項目目錄結構

3.3 項目技術選型

1）Kafka_2.11-0.11.0.0
2）Zookeeper3.4.10
3）Hadoop2.7.2
4）Flume1.7.0
5）Tomcat7.0.72
6）Mysql 5.6.24
7）SSM 框架
8）echarts.jsnode

3.4 項目總體集羣規劃

3.5 建立項目工程

File --> New --> Project --> Java --> JDK 1.8 --> Nextmysql

點擊 Nextlinux

輸入項目名稱和工做空間地址android

四 APP 數據生成模塊

4.1 建立公共模塊工程

4.1.1 建立 Java 工程，導入 pom 文件

1）建立 app_logs_common 工程，主要用於編寫公共的 javabean 和工具類。
2）添加 maven 支持，在 app_logs_common 模塊上右鍵，選擇 Add Framework Support… --> 勾選maven，而後點擊 OK。ios

3）導入 pom 文件，並刷新一下 maven

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.atguigu</groupId>
    <artifactId>app_logs_common</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <!-- 地理信息工具類 -->
        <dependency>
            <groupId>com.maxmind.db</groupId>
            <artifactId>maxmind-db</artifactId>
            <version>1.0.0</version>
        </dependency>

        <!-- 加載地理信息工具類 -->
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-core</artifactId>
            <version>4.3.4.RELEASE</version>
        </dependency>
    </dependencies>
</project>

刷新一下 maven

4.1.2 建立 AppBaseLog 基類

1）建立 package 和 AppBaseLog 基類：com.atguigu.app.common.AppBaseLog

2）編寫具體的 AppBaseLog 基類內容：

package com.atguigu.app.common;

import java.io.Serializable;

/**
 * AppBaseLog 基類
 */
public class AppBaseLog implements Serializable {
    private Long createdAtMs;           // 日誌建立時間
    private String appId;               // 應用惟一標識
    private String tenantId;            // 租戶惟一標識，企業用戶
    private String deviceId;            // 設備惟一標識
    private String appVersion;          // 版本
    private String appChannel;          // 渠道，安裝時就在清單中指定了，如 appStore 等
    private String appPlatform;         // 平臺
    private String osType;              // 操做系統
    private String deviceStyle;         // 機型

    // getter 和 setter 省略
}

4.1.3 建立 AppErrorLog 錯誤日誌類

package com.atguigu.app.common;

/**
 * 應用上報的 app 錯誤日誌相關信息
 */
public class AppErrorLog extends AppBaseLog {

    private String errorBrief;      // 錯誤摘要
    private String errorDetail;     // 錯誤詳情

    // getter 和 setter 省略
}

4.1.4 建立 AppEventLog 事件日誌類

package com.atguigu.app.common;

import java.util.Map;

/**
 * 應用上報的事件相關信息
 */
public class AppEventLog extends AppBaseLog {

    private String eventId;                         // 事件惟一標識
    private Long eventDurationSecs;                 // 事件持續時長
    private Map<String, String> paramKeyValueMap;   // 參數名/值對

    // getter 和 setter 省略
}

4.1.5 建立 AppPageLog 頁面日誌類

package com.atguigu.app.common;

/**
 * 應用上報的頁面相關信息
 */
public class AppPageLog extends AppBaseLog {

    private String pageId;                      // 頁面 id
    private int visitIndex = 0;                 // 訪問順序號，0 爲第一個頁面
    private String nextPage;                    // 下一個訪問頁面，如爲空則表示爲退出應用的頁面
    private Long stayDurationSecs = (long) 0;   // 當前頁面停留時長

    // getter 和 setter 省略
}

4.1.6 建立 AppStartupLog 啓動日誌類

package com.atguigu.app.common;

/**
 * app 啓動日誌
 */
public class AppStartupLog extends AppBaseLog {
    private String country;         // 國家，終端不用上報，服務器自動填充該屬性
    private String province;        // 省份，終端不用上報，服務器自動填充該屬性
    private String ipAddress;       // ip地址

    private String network;         // 網絡
    private String carrier;         // 運營商

    private String brand;           // 品牌
    private String screenSize;      // 分辨率

    // getter 和 setter 省略
}

4.1.7 建立 AppUsageLog 使用時長日誌類

package com.atguigu.app.common;

/**
 * 應用上報的使用時長相關信息
 */
public class AppUsageLog extends AppBaseLog {

    private Long singleUseDurationSecs;     // 單次使用時長(秒數)，指一次啓動內應用在前臺的持續時長
    private Long singleUploadTraffic;       // 單次使用過程當中的上傳流量
    private Long singleDownloadTraffic;     // 單次使用過程當中的下載流量

    // getter 和 setter 省略
}

4.1.8 建立 AppLogEntity 日誌實體類

package com.atguigu.app.common;

/**
 * AppLogEntity 日誌實體類
 * 內部含有各類日誌時間的集合
 */
public class AppLogEntity extends AppBaseLog {

    private AppStartupLog[] appStartupLogs; // 啓動相關信息的數組
    private AppPageLog[] appPageLogs;       // 頁面跳轉相關信息的數組
    private AppEventLog[] appEventLogs;     // 事件相關信息的數組
    private AppUsageLog[] appUsageLogs;     // app 使用狀況相關信息的數組
    private AppErrorLog[] appErrorLogs;     // 錯誤相關信息的數組

    // getter 和 setter 省略
}

4.1.9 建立 GeoInfo 地理信息類

package com.atguigu.app.common;

/**
 * 地理信息類
 */
public class GeoInfo {
    private String country;
    private String province;

    // getter 和 setter 省略
}

4.1.10 建立 GeoUtil 地理信息工具類

1）建立一個工具類包：com.atguigu.app.utils
2）編寫工具類：
地理信息數據庫官方地址：https://www.maxmind.com/zh/geoip2-databases
API 使用說明：http://maxmind.github.io/GeoIP2-java/

package com.atguigu.app.utils;

import com.fasterxml.jackson.databind.JsonNode;
import com.maxmind.db.Reader;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.Resource;

import java.io.IOException;
import java.io.InputStream;
import java.net.InetAddress;

/**
 * 地理工具類，實現經過 ip 查找地址區域
 */
public class GeoUtil {
    private static InputStream in;
    private static Reader reader;

    /**
     * 得到國家數據
     */
    public static String getCountry(String ip) {
        try {
            Resource resource = new ClassPathResource("GeoLite2-City.mmdb");
            reader = new Reader(resource.getFile());

            if (reader != null) {
                JsonNode node = reader.get(InetAddress.getByName(ip));

                if (node != null) {
                    JsonNode countryNode = node.get("country");

                    if (countryNode != null) {
                        JsonNode namesNode = countryNode.get("names");

                        if (namesNode != null) {
                            JsonNode zhNode = namesNode.get("zh-CN");

                            if (zhNode != null) {
                                return zhNode.textValue();
                            }
                        }
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (reader != null) {
                try {
                    reader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        return "";
    }

    /**
     * 得到省份數據
     */
    public static String getProvince(String ip) {

        try {

            Resource resource = new ClassPathResource("GeoLite2-City.mmdb");
            reader = new Reader(resource.getFile());

            if (reader != null) {
                JsonNode node = reader.get(InetAddress.getByName(ip));

                if (node != null) {
                    JsonNode subdivisionsNode = node.get("subdivisions");
                    if (subdivisionsNode != null) {
                        JsonNode areaNode = subdivisionsNode.get(0);

                        if (areaNode != null) {
                            JsonNode namesNode = areaNode.get("names");

                            if (namesNode != null) {
                                JsonNode zhNode = namesNode.get("zh-CN");

                                if (zhNode != null) {
                                    return zhNode.textValue();
                                }
                            }
                        }
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (reader != null) {
                try {
                    reader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        return "";
    }
}

4.2 編寫手機客戶端工程

4.2.1 建立 Java 工程，導入 pom 文件

1）建立 app_logs_client 模塊
2）添加 maven
3）導入 pom 文件，並刷新一下 maven

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.atguigu</groupId>
    <artifactId>app_logs_client</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>com.atguigu</groupId>
            <artifactId>app_logs_common</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>

        <!-- json 解析 -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.6</version>
        </dependency>
    </dependencies>
</project>

4.2.2 建立 GenerateData 數據生成類

1）建立 package：com.atguigu.app.client
2）編寫代碼

package com.atguigu.app.client;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.app.common.*;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;

public class GenerateData {
    // 0 建立隨機數對象
    private static Random random = new Random();

    // 1 準備基類log的屬性值
    // 1.1 功能屬性值
    private static Long[] createdAtMs = initCreatedAtMs();         // 日誌建立時間
    private static String appId = "sdk34734";                       // 應用惟一標識
    private static String[] tenantIds = {"cake"};                   // 租戶惟一標識,企業用戶
    private static String[] deviceIds = initDeviceId();             // 設備惟一標識
    private static String[] appVersions = {"3.2.1", "3.2.2"};       // 版本
    private static String[] appChannels = {"youmeng1", "youmeng2"}; // 渠道，安裝時就在清單中指定了，如 appStore 等
    private static String[] appPlatforms = {"android", "ios"};      // 平臺
    private static String[] osTypes = {"8.3", "7.1.1"};             // 操做系統
    private static String[] deviceStyles = {"iPhone 6", "iPhone 6 Plus", "紅米手機1s"}; // 機型

    // 1.1.1 初始化設備 id
    private static String[] initDeviceId() {
        String base = "device22";
        String[] result = new String[100];

        for (int i = 0; i < 100; i++) {
            result[i] = base + i + "";
        }

        return result;
    }

    // 1.1.2 初始化建立時間
    private static Long[] initCreatedAtMs() {
        Long createdAtMs = System.currentTimeMillis();
        Long[] result = new Long[11];

        for (int i = 0; i < 10; i++) {
            result[i] = createdAtMs - (long) (i * 24 * 3600 * 1000);
        }
        result[10] = createdAtMs;

        return result;
    }

    // 1.2 啓動日誌屬性值
    private static String[] countrys = {"America", "china"};                    // 國家，終端不用上報，服務器自動填充該屬性
    private static String[] provinces = {"Washington", "jiangxi", "beijing"};   // 省份，終端不用上報，服務器自動填充該屬性
    private static String[] networks = {"WiFi", "CellNetwork"};                 // 網絡
    private static String[] carriers = {"中國移動", "中國電信", "EE"};                    // 運營商
    private static String[] brands = {"三星", "華爲", "Apple", "魅族", "小米", "錘子"};   // 品牌
    private static String[] screenSizes = {"1136*640", "960*640", "480*320"};   // 分辨率

    // 1.3 事件日誌屬性值
    private static String[] eventIds = {"popMenu", "autoImport", "BookStore"};  // 事件惟一標識
    private static Long[] eventDurationSecsS = {new Long(25), new Long(67), new Long(45)};  // 事件持續時長
    static Map<String, String> map1 = new HashMap<String, String>() {
        {
            put("testparam1key", "testparam1value");
            put("testparam2key", "testparam2value");
        }
    };

    static Map<String, String> map2 = new HashMap<String, String>() {
        {
            put("testparam3key", "testparam3value");
            put("testparam4key", "testparam4value");
        }
    };
    private static Map[] paramKeyValueMapsS = {map1, map2}; // 參數名/值對


    // 1.4 使用時長日誌屬性值
    private static Long[] singleUseDurationSecsS = initSingleUseDurationSecs(); // 單次使用時長(秒數)，指一次啓動內應用在前臺的持續時長

    // 1.4.1 單次使用時長
    private static Long[] initSingleUseDurationSecs() {
        Long[] result = new Long[200];

        for (int i = 1; i < 200; i++) {
            result[i] = (long) random.nextInt(200);
        }

        return result;
    }

    // 1.5 錯誤日誌屬性值
    private static String[] errorBriefs = {"at cn.lift.dfdf.web.AbstractBaseController.validInbound(AbstractBaseController.java:72)", "at cn.lift" +
            ".appIn.control.CommandUtil.getInfo(CommandUtil.java:67)"}; // 錯誤摘要
    private static String[] errorDetails = {"java.lang.NullPointerException\\n " + "at cn.lift.appIn.web.AbstractBaseController.validInbound" +
            "(AbstractBaseController.java:72)\\n " + "at cn.lift.dfdf.web.AbstractBaseController.validInbound", "at cn.lift.dfdfdf.control" +
            ".CommandUtil.getInfo(CommandUtil.java:67)\\n " + "at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl" +
            ".java:43)\\n" + " at java.lang.reflect.Method.invoke(Method.java:606)\\n"}; // 錯誤詳情

    // 1.6 頁面使用狀況日誌屬性值
    private static String[] pageIds = {"list.html", "main.html", "test.html"};          // 頁面 id
    private static int[] visitIndexs = {0, 1, 2, 3, 4};                                 // 訪問順序號，0爲第一個頁面
    private static String[] nextPages = {"list.html", "main.html", "test.html", null};  // 下一個訪問頁面，如爲空則表示爲退出應用的頁面
    private static Long[] stayDurationSecsS = {new Long(45), new Long(2), new Long(78)};// 當前頁面停留時長

    // 2 初始化五類 log 的數據
    // 啓動相關信息的數組
    private static AppStartupLog[] appStartupLogs = initAppStartupLogs();
    // 頁面跳轉相關信息的數組
    private static AppPageLog[] appPageLogs = initAppPageLogs();
    // 事件相關信息的數組
    private static AppEventLog[] appEventLogs = initAppEventLogs();
    // app 使用狀況相關信息的數組
    private static AppUsageLog[] appUsageLogs = initAppUsageLogs();
    // 錯誤相關信息的數組
    private static AppErrorLog[] appErrorLogs = initAppErrorLogs();

    // 2.1 初始化每類 log 的公共屬性值
    private static void initLogCommon(AppBaseLog baselog) {
        // 日誌建立時間
        baselog.setCreatedAtMs(System.currentTimeMillis());
        // appid
        baselog.setAppId(appId);
        // 租戶惟一標識，企業用戶
        String tenantId = tenantIds[random.nextInt(tenantIds.length)];
        if (tenantId != null) {
            baselog.setTenantId(tenantId);
        }
        baselog.setTenantId(tenantIds[random.nextInt(tenantIds.length)]);
        // 設備惟一標識
        baselog.setDeviceId(deviceIds[random.nextInt(deviceIds.length)]);
        // 版本
        baselog.setAppVersion(appVersions[random.nextInt(appVersions.length)]);
        // 渠道
        baselog.setAppChannel(appChannels[random.nextInt(appChannels.length)]);
        // 平臺
        baselog.setAppPlatform(appPlatforms[random.nextInt(appPlatforms.length)]);
        // 操做系統
        baselog.setOsType(osTypes[random.nextInt(osTypes.length)]);
        // 機型
        baselog.setDeviceStyle(deviceStyles[random.nextInt(deviceStyles.length)]);
    }

    // 2.2 啓動相關信息的數組
    private static AppStartupLog[] initAppStartupLogs() {
        AppStartupLog[] result = new AppStartupLog[10];

        for (int i = 0; i < 10; i++) {
            AppStartupLog appStartupLog = new AppStartupLog();

            // 初始化公共屬性值
            initLogCommon(appStartupLog);

            // 國家
            appStartupLog.setCountry(countrys[random.nextInt(countrys.length)]);
            // 省份
            appStartupLog.setProvince(provinces[random.nextInt(provinces.length)]);
            // 網絡
            appStartupLog.setNetwork(networks[random.nextInt(networks.length)]);
            // 運營商
            appStartupLog.setCarrier(carriers[random.nextInt(carriers.length)]);
            // 品牌
            appStartupLog.setBrand(brands[random.nextInt(brands.length)]);
            // 分辨率
            appStartupLog.setScreenSize(screenSizes[random.nextInt(screenSizes.length)]);

            result[i] = appStartupLog;
        }

        return result;
    }

    // 2.3 頁面跳轉相關信息的數組
    private static AppPageLog[] initAppPageLogs() {
        AppPageLog[] result = new AppPageLog[10];

        for (int i = 0; i < 10; i++) {

            AppPageLog appPageLog = new AppPageLog();

            // 初始化公共屬性值
            initLogCommon(appPageLog);

            // 頁面 id
            String pageId = pageIds[random.nextInt(pageIds.length)];
            appPageLog.setPageId(pageId);

            // 訪問頁面順序號
            int visitIndex = visitIndexs[random.nextInt(visitIndexs.length)];
            appPageLog.setVisitIndex(visitIndex);

            // 下一個訪問頁面，如爲空則表示爲退出應用的頁面
            String nextPage = nextPages[random.nextInt(nextPages.length)];
            while (pageId.equals(nextPage)) {
                nextPage = nextPages[random.nextInt(nextPages.length)];
            }
            appPageLog.setNextPage(nextPage);

            // 當前頁面停留時長
            Long stayDurationSecs = stayDurationSecsS[random.nextInt(stayDurationSecsS.length)];
            appPageLog.setStayDurationSecs(stayDurationSecs);

            result[i] = appPageLog;
        }

        return result;
    }

    // 2.4 事件相關信息的數組
    private static AppEventLog[] initAppEventLogs() {
        AppEventLog[] result = new AppEventLog[10];

        for (int i = 0; i < 10; i++) {
            AppEventLog appEventLog = new AppEventLog();

            // 初始化公共屬性值
            initLogCommon(appEventLog);

            // 事件惟一標識
            appEventLog.setEventId(eventIds[random.nextInt(eventIds.length)]);
            // 事件持續時長
            appEventLog.setEventDurationSecs(eventDurationSecsS[random.nextInt(eventDurationSecsS.length)]);
            // 事件參數
            appEventLog.setParamKeyValueMap(paramKeyValueMapsS[random.nextInt(paramKeyValueMapsS.length)]);

            result[i] = appEventLog;
        }

        return result;
    }


    // 2.5 app 使用狀況相關信息的數組
    private static AppUsageLog[] initAppUsageLogs() {
        AppUsageLog[] result = new AppUsageLog[10];

        for (int i = 0; i < 10; i++) {
            AppUsageLog appUsageLog = new AppUsageLog();

            // 初始化公共屬性值
            initLogCommon(appUsageLog);

            // 單次使用時長(秒數)，指一次啓動內應用在前臺的持續時長
            appUsageLog.setSingleUseDurationSecs(singleUseDurationSecsS[random.nextInt(singleUseDurationSecsS.length)]);

            result[i] = appUsageLog;
        }

        return result;
    }

    // 2.6 錯誤相關信息的數組
    private static AppErrorLog[] initAppErrorLogs() {
        AppErrorLog[] result = new AppErrorLog[10];

        for (int i = 0; i < 10; i++) {
            AppErrorLog appErrorLog = new AppErrorLog();

            initLogCommon(appErrorLog);

            // 錯誤摘要
            appErrorLog.setErrorBrief(errorBriefs[random.nextInt(errorBriefs.length)]);
            // 錯誤詳情
            appErrorLog.setErrorDetail(errorDetails[random.nextInt(errorDetails.length)]);

            result[i] = appErrorLog;
        }

        return result;
    }

    // 3 循環發送數據
    public static void main(String[] args) {
        // 發送數據
        for (int i = 1; i <= 200000000; i++) {
            AppLogEntity logEntity = new AppLogEntity();

            // 封裝5 種 log 數據
            logEntity.setAppStartupLogs(new AppStartupLog[]{appStartupLogs[random.nextInt(appStartupLogs.length)]});
            logEntity.setAppEventLogs(new AppEventLog[]{appEventLogs[random.nextInt(appEventLogs.length)]});
            logEntity.setAppErrorLogs(new AppErrorLog[]{appErrorLogs[random.nextInt(appErrorLogs.length)]});
            logEntity.setAppPageLogs(new AppPageLog[]{appPageLogs[random.nextInt(appPageLogs.length)]});
            logEntity.setAppUsageLogs(new AppUsageLog[]{appUsageLogs[random.nextInt(appUsageLogs.length)]});

            try {
                // 將對象轉換成 json string
                String json = JSONObject.toJSONString(logEntity);

                // 網絡請求發送 json 數據
                UploadUtil.upload(json);

                // 每隔 5 秒發送一條數據
                Thread.sleep(5000);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

4.2.3 建立數據上傳的工具類

package com.atguigu.app.client;

import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;

/**
 * 數據上傳的工具類
 */
public class UploadUtil {
    /**
     * 上傳日誌
     */
    public static void upload(String json) throws Exception {
        try {
            // 1 設置請求的 URL
            // URL url = new URL("http://hadoop102:8080/app_logs/coll/index"); // 生產地址
            URL url = new URL("http://localhost:8080/coll/index"); // 測試地址

            // 2 獲取鏈接
            HttpURLConnection conn = (HttpURLConnection) url.openConnection();

            // 2.1 設置請求方式爲 post
            conn.setRequestMethod("POST");

            // 2.2 容許上傳數據使能
            conn.setDoOutput(true);

            // 2.3 時間頭用來供 server 進行時鐘校對
            conn.setRequestProperty("clientTime", System.currentTimeMillis() + "");

            // 2.4 設置請求的頭信息，設置內容類型
            conn.setRequestProperty("Content-Type", "application/json");

            // 3 獲取輸出流
            OutputStream out = conn.getOutputStream();
            // 3.1 向輸出流裏面寫數據
            out.write(json.getBytes());
            out.flush();
            // 3.2 關閉資源
            out.close();

            // 4 獲取響應碼
            int code = conn.getResponseCode();
            System.out.println(code);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

五數據收集模塊

5.1 Web 數據收集模塊

5.1.0 數據收集模塊集羣部署規劃

5.1.1 建立 web 工程，導入 pom 文件

1）建立 web 工程：app_logs_collect_web，輸入項目名稱和工做空間地址，點擊完成便可。

2）添加 maven 工具
3）導入 pom 文件，並刷新一下 maven

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.atguigu</groupId>
    <artifactId>app_logs_collect_web</artifactId>
    <version>1.0-SNAPSHOT</version>

    <packaging>war</packaging>

    <!-- 日誌框架版本號 -->
    <properties>
        <log4j.version>1.2.17</log4j.version>
        <slf4j.version>1.7.22</slf4j.version>
    </properties>

    <dependencies>
        <!-- spring 框架 -->
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context-support</artifactId>
            <version>4.3.4.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-webmvc</artifactId>
            <version>4.3.4.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>commons-io</groupId>
            <artifactId>commons-io</artifactId>
            <version>2.5</version>
        </dependency>

        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-pool2</artifactId>
            <version>2.4.2</version>
        </dependency>
        <dependency>
            <groupId>javax.servlet</groupId>
            <artifactId>javax.servlet-api</artifactId>
            <version>3.0.1</version>
            <scope>provided</scope>
        </dependency>

        <!-- kafka 框架 -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.10.2.1</version>
        </dependency>

        <!-- 打印日誌框架 -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>jcl-over-slf4j</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>${log4j.version}</version>
        </dependency>

        <!-- jackson json 解析框架-->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-core</artifactId>
            <version>2.8.8</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.8.3</version>
        </dependency>

        <!-- fastjson json 解析框架-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.24</version>
        </dependency>

        <!-- 項目公共模塊 -->
        <dependency>
            <groupId>com.atguigu</groupId>
            <artifactId>app_logs_common</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>

        <!-- 加載地理信息 -->
        <dependency>
            <groupId>com.maxmind.db</groupId>
            <artifactId>maxmind-db</artifactId>
            <version>1.0.0</version>
        </dependency>
    </dependencies>

    <!-- tomcat 插件-->
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.tomcat.maven</groupId>
                <artifactId>tomcat7-maven-plugin</artifactId>
                <version>2.2</version>
                <configuration>
                    <!-- http port -->
                    <port>8080</port>
                    <path>/</path>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

5.1.2 在 web.xml 文件中加載 Spring 和 Springmvc 配置

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://xmlns.jcp.org/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/web-app_3_1.xsd"
         version="3.1">

    <!-- 1 spring 加載配置 -->
    <listener>
        <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
    </listener>
    <context-param>
        <param-name>contextConfigLocation</param-name>
        <param-value>classpath:beans.xml</param-value>
    </context-param>

    <!-- 2 springmvc 加載配置 -->
    <servlet>
        <servlet-name>dispatcher</servlet-name>
        <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
        <init-param>
            <param-name>contextConfigLocation</param-name>
            <param-value>classpath:dispatcher-servlet.xml</param-value>
        </init-param>
        <load-on-startup>1</load-on-startup>
    </servlet>

    <servlet-mapping>
        <servlet-name>dispatcher</servlet-name>
        <url-pattern>/</url-pattern>
    </servlet-mapping>

    <!-- 3 解決傳輸中文亂碼 -->
    <filter>
        <filter-name>CharacterEncodingFilter</filter-name>
        <filter-class>org.springframework.web.filter.CharacterEncodingFilter</filter-class>
        <init-param>
            <param-name>encoding</param-name>
            <param-value>UTF-8</param-value>
        </init-param>
        <init-param>
            <param-name>forceEncoding</param-name>
            <param-value>true</param-value>
        </init-param>
    </filter>
    <filter-mapping>
        <filter-name>CharacterEncodingFilter</filter-name>
        <url-pattern>/*</url-pattern>
    </filter-mapping>
</web-app>

5.1.3 在 resources 路徑下添加 Springmvc 配置文件

1）dispatcher-servlet.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:util="http://www.springframework.org/schema/util"
       xmlns:aop="http://www.springframework.org/schema/aop"
       xmlns:p="http://www.springframework.org/schema/p"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:mvc="http://www.springframework.org/schema/mvc"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
                           http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-3.2.xsd
                           http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.2.xsd
                           http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.2.xsd
                           http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-3.2.xsd">

    <!-- 配置掃描路徑 -->
    <context:component-scan base-package="com.atguigu.applogs.collect.web.controller"/>

    <!-- 使用註解驅動 -->
    <mvc:annotation-driven/>

    <!-- 內部資源視圖解析器 -->
    <bean class="org.springframework.web.servlet.view.InternalResourceViewResolver">
        <property name="prefix" value="/"/>
        <property name="suffix" value=".jsp"/>
    </bean>
</beans>

2）建立包名：com.atguigu.applogs.collect.web.controller

5.1.4 在 resources 路徑下添加 Spring 配置文件

beans.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:util="http://www.springframework.org/schema/util"
       xmlns:aop="http://www.springframework.org/schema/aop"
       xmlns:p="http://www.springframework.org/schema/p"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
                           http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
                           http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd
                           http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
                           ">
    <!--配置路徑掃描-->
    <context:component-scan base-package="com.atguigu.applogs.collect.web"/>
</beans>

5.1.5 在 resources 路徑下添加 log4j 文件

log4j.properties

log4j.rootLogger=info, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS}  %5p --- [%50t]  %-80c(line:%5L)  :  %m%n

log4j.appender.R=org.apache.log4j.RollingFileAppender
log4j.appender.R.File=analysis.log
log4j.appender.R.MaxFileSize=1024KB
log4j.appender.R.MaxBackupIndex=1

log4j.appender.R.layout=org.apache.log4j.PatternLayout
log4j.appender.R.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS}  %5p --- [%50t]  %-80c(line:%6L)  :  %m%n

5.1.6 網絡請求處理

編寫代碼

package com.atguigu.applogs.collect.web.controller;

import com.atguigu.app.common.AppLogEntity;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;

import javax.servlet.http.HttpServletRequest;

@Controller
@RequestMapping("/coll")
public class CollectLogController {

    @RequestMapping(value = "/index", method = RequestMethod.POST)
    @ResponseBody
    public AppLogEntity collect(@RequestBody AppLogEntity e, HttpServletRequest req) {

        // 1 修正服務器和客戶端時間
        verifyTime(e, req);

        // 2 獲取國家、省份和 ip 地址信息
        processIp(e, req);

        // 3 向 Kafka 發送消息
        sendMessage(e);

        return e;
    }
}

5.1.7 修正服務器和客戶端時間

1）分析

2）編碼實現

    // 修正時間
    private void verifyTime(AppLogEntity e, HttpServletRequest req) {

        // 1 獲取服務器時間
        long myTime = System.currentTimeMillis();
        // 2 獲取客戶端時間
        long clientTime = Long.parseLong(req.getHeader("clientTime"));
        // 3 計算服務器和客戶端時間差
        long diff = myTime - clientTime;

        // 4 根據時間差，修正日誌中時間
        for (AppStartupLog log : e.getAppStartupLogs()) {
            log.setCreatedAtMs(log.getCreatedAtMs() + diff);
        }

        for (AppUsageLog log : e.getAppUsageLogs()) {
            log.setCreatedAtMs(log.getCreatedAtMs() + diff);
        }

        for (AppPageLog log : e.getAppPageLogs()) {
            log.setCreatedAtMs(log.getCreatedAtMs() + diff);
        }

        for (AppEventLog log : e.getAppEventLogs()) {
            log.setCreatedAtMs(log.getCreatedAtMs() + diff);
        }

        for (AppErrorLog log : e.getAppErrorLogs()) {
            log.setCreatedAtMs(log.getCreatedAtMs() + diff);
        }
    }

5.1.8 獲取國家、省份、和IP地址信息

1）在 resource 路徑下添加 GeoLite2-City.mmdb 資源
2）編碼實現
根據 ip 地址查詢國家和省份信息，並作緩存處理。

    // 處理 ip client 地址問題
    private void processIp(AppLogEntity e, HttpServletRequest req) {

        // 1 獲取客戶端 ip 地址
        String clientIP = req.getRemoteAddr();

        // 2 從緩存中獲取數據
        GeoInfo geoInfo = cache.get(clientIP);

        // 若是該客戶端 ip 地址沒有獲取過國家和省份信息，則經過工具類獲取
        // 若是該客戶端 ip 地址已經獲取過國家和省份信息，則直接從緩存對象中獲取
        if (geoInfo == null) {
            geoInfo = new GeoInfo();
            geoInfo.setCountry(GeoUtil.getCountry(clientIP));
            geoInfo.setProvince(GeoUtil.getProvince(clientIP));

            // 緩存數據
            cache.put(clientIP, geoInfo);
        }

        // 3 設置國家、省份和客戶端 ip 地址信息
        for (AppStartupLog log : e.getAppStartupLogs()) {
            log.setCountry(geoInfo.getCountry());
            log.setProvince(geoInfo.getProvince());
            log.setIpAddress(clientIP);
        }
    }

緩存代碼

    // 緩存地址信息
    private Map<String, GeoInfo> cache = new HashMap<String, GeoInfo>();

5.1.9 向 Kafka 發送消息

1）在 app_logs_common 模塊中在 com.atguigu.app.common 包下添加常量（5個 topic 主題）

package com.atguigu.app.common;

/**
 * 常量類
 */
public class Constants {
    // 主題
    public static final String TOPIC_APP_STARTUP = "topic_app_startup";
    public static final String TOPIC_APP_ERRROR = "topic_app_error";
    public static final String TOPIC_APP_EVENT = "topic_app_event";
    public static final String TOPIC_APP_USAGE = "topic_app_usage";
    public static final String TOPIC_APP_PAGE = "topic_app_page";
}

2）編寫代碼（app_logs_collect_web 模塊）

    // 發送消息給發 Kafka
    private void sendMessage(AppLogEntity e) {
        // 1 建立配置對象
        Properties props = new Properties();
        // 1.1 Kafka 服務端的主機名和端口號
        props.put("bootstrap.servers", "hadoop102:9092");
        // 1.2 等待全部副本節點的應答
        props.put("acks", "all");
        // 1.3 消息發送最大嘗試次數
        props.put("retries", 0);
        // 1.4 一批消息處理大小
        props.put("batch.size", 16384);
        // 1.5 請求延時
        props.put("linger.ms", 1);
        // 1.6 發送緩存區內存大小
        props.put("buffer.memory", 33554432);
        // 1.7 key 序列化
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 1.8 value 序列化
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 2 建立生產者
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

        // 3 根據日誌類型分別向 5 個主題發送消息
        sendSingleLog(producer, Constants.TOPIC_APP_STARTUP, e.getAppStartupLogs());
        sendSingleLog(producer, Constants.TOPIC_APP_ERRROR, e.getAppErrorLogs());
        sendSingleLog(producer, Constants.TOPIC_APP_EVENT, e.getAppEventLogs());
        sendSingleLog(producer, Constants.TOPIC_APP_PAGE, e.getAppPageLogs());
        sendSingleLog(producer, Constants.TOPIC_APP_USAGE, e.getAppUsageLogs());

        // 4 關閉生產者
        producer.close();
    }

    // 發送單個的 log 消息給 kafka
    private void sendSingleLog(KafkaProducer<String, String> producer, String topic, AppBaseLog[] logs) {
        for (AppBaseLog log : logs) {
            // 1 將 bean 對象轉換爲 json
            String logMsg = JSONObject.toJSONString(log);

            // 2 建立待發送消息對象
            ProducerRecord<String, String> data = new ProducerRecord<String, String>(topic, logMsg);

            // 3 發送消息
            producer.send(data);
        }
    }

完整示例代碼以下：
com\atguigu\applogs\collect\web\controller\CollectLogController.java

package com.atguigu.applogs.collect.web.controller;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.app.common.*;
import com.atguigu.app.utils.GeoUtil;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;

import javax.servlet.http.HttpServletRequest;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

@Controller
@RequestMapping("/coll")
public class CollectLogController {

    // 緩存地址信息
    private Map<String, GeoInfo> cache = new HashMap<String, GeoInfo>();

    @RequestMapping(value = "/index", method = RequestMethod.POST)
    @ResponseBody
    public AppLogEntity collect(@RequestBody AppLogEntity e, HttpServletRequest req) {
        // 1 修正服務器和客戶端時間
        verifyTime(e, req);

        // 2 獲取國家、省份和 ip 地址信息
        processIp(e, req);

        // 3 向 Kafka 發送消息
        sendMessage(e);

        return e;
    }

    // 修正時間
    private void verifyTime(AppLogEntity e, HttpServletRequest req) {
        // 1 獲取服務器時間
        long myTime = System.currentTimeMillis();
        // 2 獲取客戶端時間
        long clientTime = Long.parseLong(req.getHeader("clientTime"));
        // 3 計算服務器和客戶端時間差
        long diff = myTime - clientTime;

        // 4 根據時間差，修正日誌中時間
        for (AppStartupLog log : e.getAppStartupLogs()) {
            log.setCreatedAtMs(log.getCreatedAtMs() + diff);
        }

        for (AppUsageLog log : e.getAppUsageLogs()) {
            log.setCreatedAtMs(log.getCreatedAtMs() + diff);
        }

        for (AppPageLog log : e.getAppPageLogs()) {
            log.setCreatedAtMs(log.getCreatedAtMs() + diff);
        }

        for (AppEventLog log : e.getAppEventLogs()) {
            log.setCreatedAtMs(log.getCreatedAtMs() + diff);
        }

        for (AppErrorLog log : e.getAppErrorLogs()) {
            log.setCreatedAtMs(log.getCreatedAtMs() + diff);
        }
    }

    // 處理 ip client 地址問題
    private void processIp(AppLogEntity e, HttpServletRequest req) {
        // 1 獲取客戶端 ip 地址
        String clientIP = req.getRemoteAddr();

        // 2 從緩存中獲取數據
        GeoInfo geoInfo = cache.get(clientIP);

        // 若是該客戶端 ip 地址沒有獲取過國家和省份信息，則經過工具類獲取
        // 若是該客戶端 ip 地址已經獲取過國家和省份信息，則直接從緩存對象中獲取
        if (geoInfo == null) {
            geoInfo = new GeoInfo();
            geoInfo.setCountry(GeoUtil.getCountry(clientIP));
            geoInfo.setProvince(GeoUtil.getProvince(clientIP));

            // 緩存數據
            cache.put(clientIP, geoInfo);
        }

        // 3 設置國家、省份和客戶端 ip 地址信息
        for (AppStartupLog log : e.getAppStartupLogs()) {
            log.setCountry(geoInfo.getCountry());
            log.setProvince(geoInfo.getProvince());
            log.setIpAddress(clientIP);
        }
    }

    // 發送消息給發 Kafka
    private void sendMessage(AppLogEntity e) {
        // 1 建立配置對象
        Properties props = new Properties();
        // 1.1 Kafka 服務端的主機名和端口號
        props.put("bootstrap.servers", "hadoop102:9092");
        // 1.2 等待全部副本節點的應答
        props.put("acks", "all");
        // 1.3 消息發送最大嘗試次數
        props.put("retries", 0);
        // 1.4 一批消息處理大小
        props.put("batch.size", 16384);
        // 1.5 請求延時
        props.put("linger.ms", 1);
        // 1.6 發送緩存區內存大小
        props.put("buffer.memory", 33554432);
        // 1.7 key 序列化
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 1.8 value 序列化
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 2 建立生產者
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

        // 3 根據日誌類型分別向 5 個主題發送消息
        sendSingleLog(producer, Constants.TOPIC_APP_STARTUP, e.getAppStartupLogs());
        sendSingleLog(producer, Constants.TOPIC_APP_ERRROR, e.getAppErrorLogs());
        sendSingleLog(producer, Constants.TOPIC_APP_EVENT, e.getAppEventLogs());
        sendSingleLog(producer, Constants.TOPIC_APP_PAGE, e.getAppPageLogs());
        sendSingleLog(producer, Constants.TOPIC_APP_USAGE, e.getAppUsageLogs());

        // 4 關閉生產者
        producer.close();
    }

    // 發送單個的 log 消息給 kafka
    private void sendSingleLog(KafkaProducer<String, String> producer, String topic, AppBaseLog[] logs) {
        for (AppBaseLog log : logs) {
            // 1 將 bean 對象轉換爲 json
            String logMsg = JSONObject.toJSONString(log);

            // 2 建立待發送消息對象
            ProducerRecord<String, String> data = new ProducerRecord<String, String>(topic, logMsg);

            // 3 發送消息
            producer.send(data);
        }
    }
}

5.1.10 在 IDEA 上執行 Web 程序（測試環境）

0）因爲目前咱們 Kafka 集羣尚未配置出來，因此咱們向 Kafka 發送消息，並不能獲得 Kafka 的及時應答，客戶端因爲收不到應答，就會卡死在那，因此爲了測試方便，咱們先將向 Kafka 發送消息註釋掉，以下：

1）調整 web 項目結構，將 web 文件夾總體拖拽到 main 文件夾下，並修更名稱爲 webapp。不然，後面將該 web 包打包的時候會報錯！！！

2）配置 web 執行程序方式
① 點擊 Edit Configurations…

② 點擊綠色的加號 --> 找到 Tomcat Server --> Local

③ 點擊 Deployment --> 綠色加號 --> Artifact…

④ 選擇帶有 exploded 的哪項

⑤ 輸入 Application Servers 名稱：app_logs_collect_web，再依次按照圖示選擇

3）向 web 工程中添加 jar 包依賴

5.1.11 測試

1）啓動 tomact7

瀏覽器訪問：http://localhost:8080/

2）啓動客戶端日誌生成程序 app_statistics_project\app_logs_client\src\main\java\com\atguigu\app\client\GenerateData.java，查看是否正確收到數據響應碼 200。

5.2 Kafka 集羣模塊

5.2.1 配置 Zookeeper 集羣

1）具體配置詳見：http://www.javashuo.com/article/p-gbhycygo-kb.html
2）先啓動 zookeeper 集羣

[atguigu@hadoop102 zookeeper-3.4.10]$ bin/zkServer.sh start
[atguigu@hadoop103 zookeeper-3.4.10]$ bin/zkServer.sh start
[atguigu@hadoop104 zookeeper-3.4.10]$ bin/zkServer.sh start

5.2.2 配置 Kafka 集羣

0）具體配置詳見：http://www.javashuo.com/article/p-sewqjxhc-cz.html
1）再啓動 kafka 集羣

[atguigu@hadoop102 kafka]$ bin/kafka-server-start.sh config/server.properties &
[atguigu@hadoop103 kafka]$ bin/kafka-server-start.sh config/server.properties &
[atguigu@hadoop104 kafka]$ bin/kafka-server-start.sh config/server.properties &

2）查看主題 topic

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --zookeeper hadoop102:2181 --list

3）建立 kafka 的 topic

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --create --zookeeper hadoop102:2181 --create --replication-factor 3 --partitions 1 --topic topic_app_startup
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --create --zookeeper hadoop102:2181 --create --replication-factor 3 --partitions 1 --topic topic_app_error
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --create --zookeeper hadoop102:2181 --create --replication-factor 3 --partitions 1 --topic topic_app_event
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --create --zookeeper hadoop102:2181 --create --replication-factor 3 --partitions 1 --topic topic_app_usage
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --create --zookeeper hadoop102:2181 --create --replication-factor 3 --partitions 1 --topic topic_app_page

5.2.3 測試

1）建立消費者主題（主要用於測試數據是否可以接收到）

[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --zookeeper hadoop102:2181 --topic topic_app_startup

2）啓動客戶端日誌生成程序。
注意：此時咱們已經配置好 Kafka 集羣了，須要將向 Kafka 發送消息的代碼註釋取消，
代碼位置：D:\learn\JetBrains\workspace_idea3\app_statistics_project\app_logs_collect_web\src\main\java\com\atguigu\applogs\collect\web\controller\CollectLogController.java）
3）觀察 kafka 消費者是否正常消費到消息。

5.2.4 在 Centos 上部署 Tomcat（生產環境）

1）在 linux 上安裝 Tomcat
（1）將 apache-tomcat-7.0.72.tar.gz 導入到 linux 的 /opt/software 路徑下
（2）解壓 apache-tomcat-7.0.72.tar.gz 到 /opt/module/ 路徑下

[atguigu@hadoop102 software]$ tar -zxf apache-tomcat-7.0.72.tar.gz -C /opt/module/

（3）修更名稱爲 tomcat

[atguigu@hadoop102 module]$ mv apache-tomcat-7.0.72 tomcat

2）部署 Web 程序到 Tomcat
（1）將 web 工程(app_logs_collect_web) 打成 war 包，名稱爲app_logs.war，步驟以下：
a) 檢查 pom 文件中是否添加，<packaging>war</packaging>
b) 在 maven project 中找到 app_logs_collect_web 工程，點擊 clean，而後點擊 package。生成 war 包。

c) 查找到 war 包，並修更名稱爲 app_logs。

修改 app_logs_collect_web-1.0-SNAPSHOT.war 的名稱爲 app_logs.war

（2）將 app_logs.war 導入到 linux 的 /opt/module/tomcat/webapps 路徑下

（3）啓動 Tomcat

[atguigu@hadoop102 bin]$ pwd
/opt/module/tomcat/bin
[atguigu@hadoop102 bin]$ ./startup.sh

5.2.5 測試

修改 app_logs_client 工程中的 UploadUtil.java 中的請求地址爲：

URL url = new URL("http://hadoop102:8080/app_logs/coll/index");// 生產地址

執行日誌生成程序，觀察可否正確收到響應碼 200，不然須要進一步調試。

5.3 Flume 模塊

5.3.1 配置 Hadoop 集羣

1）配置 hadoop 集羣，詳見：http://www.javashuo.com/article/p-uyrkkoeu-gt.html
2）啓動 hadoop 集羣

[atguigu@hadoop102 hadoop-2.7.2]$ sbin/start-dfs.sh
[atguigu@hadoop103 hadoop-2.7.2]$ sbin/start-yarn.sh

5.3.2 Flume 安裝

1）將 apache-flume-1.7.0-bin.tar.gz 導入到 Linux 系統中 /opt/sotfware 目錄下
2）解壓 apache-flume-1.7.0-bin.tar.gz 到 /opt/module 目錄下

[atguigu@hadoop102 software]$ tar -zxf apache-flume-1.7.0-bin.tar.gz -C /opt/module/

3）修改 apache-flume-1.7.0-bin 名稱爲 flume

[atguigu@hadoop102 module]$ mv apache-flume-1.7.0-bin/ flume

4）使用 root 用戶，配置 flume 環境變量

[root@hadoop102 flume]# vim /etc/profile
#FLUME_HOME
export FLUME_HOME=/opt/module/flume
export PATH=$PATH:$FLUME_HOME/bin

[root@hadoop102 flume]# source /etc/profile

5）驗證 flume 環境變量配置

[root@hadoop102 flume]# flume-ng version
Flume 1.7.0

5.3.3 建立 Flume 攔截器

1）建立攔截器主要目的是區分 kafka 傳遞過來的日誌類型。
2）自定義攔截器實操
（1）建立 Java 工程 app_logs_flume
（2）添加 maven 框架支持，並導入 pom 文件，並刷新一下 maven

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.atguigu</groupId>
    <artifactId>app_logs_flume</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.apache.flume</groupId>
            <artifactId>flume-ng-core</artifactId>
            <version>1.7.0</version>
        </dependency>
    </dependencies>
</project>

2）代碼實現
a) 建立包名：com.atguigu.app.flume.interceptor
b) 根據系統時間攔截器自定義攔截器：Ctrl + Shift + t，輸入TimestampInterceptor，將系統時間攔截器中代碼拷貝過來，替換 TimestampInterceptor 的名稱爲 LogCollInterceptor。
c) 編寫鏈接器業務代碼

package com.atguigu.app.flume.interceptor;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

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

import static org.apache.flume.interceptor.TimestampInterceptor.Constants.*;

/**
 * 自定義 flume 的攔截器，提取 body 中的日誌類型做爲header
 */
public class LogCollInterceptor implements Interceptor {

    private final boolean preserveExisting;

    private LogCollInterceptor(boolean preserveExisting) {
        this.preserveExisting = preserveExisting;
    }

    public void initialize() {

    }

    /**
     * Modifies events in-place.
     */
    public Event intercept(Event event) {
        // 1 獲取 flume 接收消息頭
        Map<String, String> headers = event.getHeaders();

        // 2 獲取 flume 接收的 json 數據數組
        byte[] json = event.getBody();
        // 將 json 數組轉換爲字符串
        String jsonStr = new String(json);

        // pageLog
        String logType = "";
        if (jsonStr.contains("pageId")) {
            logType = "page";
        }
        // eventLog
        else if (jsonStr.contains("eventId")) {
            logType = "event";
        }
        // usageLog
        else if (jsonStr.contains("singleUseDurationSecs")) {
            logType = "usage";
        }
        // error
        else if (jsonStr.contains("errorBrief")) {
            logType = "error";
        }
        // startup
        else if (jsonStr.contains("network")) {
            logType = "startup";
        }

        // 3 將日誌類型存儲到 flume 頭中
        headers.put("logType", logType);

        return event;
    }

    /**
     * Delegates to {@link #intercept(Event)} in a loop.
     *
     * @param events
     * @return
     */
    public List<Event> intercept(List<Event> events) {
        for (Event event : events) {
            intercept(event);
        }
        return events;
    }

    public void close() {
    }

    public static class Builder implements Interceptor.Builder {

        private boolean preserveExisting = PRESERVE_DFLT;

        public Interceptor build() {
            return new LogCollInterceptor(preserveExisting);
        }

        public void configure(Context context) {
            preserveExisting = context.getBoolean(PRESERVE, PRESERVE_DFLT);
        }
    }

    public static class Constants {
        public static String TIMESTAMP = "timestamp";
        public static String PRESERVE = "preserveExisting";
        public static boolean PRESERVE_DFLT = false;
    }
}

（3）打成 jar 包，修改 jar 包名稱爲 app_logs_flume.jar，而後將該 jar 包導入 linux 虛擬機 /opt/module/flume/lib 目錄下。

5.3.4 配置 Flume

1）配置需求：實現消費 kafka 的 5 個主題，並把數據導入到 HDFS 文件系統。
2）配置實現：注意本博主將 flume 的配置文件放到自定義的目錄中 /opt/module/flume/job/app_logs

[atguigu@hadoop102 app_logs]$ pwd
/opt/module/flume/job/app_logs
[atguigu@hadoop102 app_logs]$ vim kafka-flume-hdfs.conf

打開文件，添加如下內容：

a1.sources = r1
a1.channels = c1
a1.sinks = k1

# 新加內容 flume 攔截器
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.atguigu.app.flume.interceptor.LogCollInterceptor$Builder
a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource

a1.sources.r1.batchSize = 5000
a1.sources.r1.batchDurationMillis = 2000
a1.sources.r1.kafka.bootstrap.servers = hadoop102:9092
a1.sources.r1.kafka.zookeeperConnect = hadoop102:2181,hadoop103:2181,hadoop104:2181
a1.sources.r1.kafka.topics=topic_app_startup,topic_app_error,topic_app_event,topic_app_usage,topic_app_page

a1.channels.c1.type = memory
a1.channels.c1.capacity = 100000
a1.channels.c1.transactionCapacity = 10000

a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /user/centos/applogs/%{logType}/%Y%m/%d/%H%M
a1.sinks.k1.hdfs.filePrefix = events-
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 30
a1.sinks.k1.hdfs.roundUnit = second

#不要產生大量小文件
a1.sinks.k1.hdfs.rollInterval = 30
a1.sinks.k1.hdfs.rollSize = 0
a1.sinks.k1.hdfs.rollCount = 0

#控制輸出文件是原生文件
a1.sinks.k1.hdfs.fileType = DataStream
a1.sources.r1.channels = c1
a1.sinks.k1.channel= c1

3）啓動 flume

[atguigu@hadoop102 conf]$ bin/flume-ng agent --conf conf/ --name a1 --conf-file job/app_logs/kafka-flume-hdfs.conf -Dflume.root.logger=INFO,console  或者
[atguigu@hadoop102 conf]$ bin/flume-ng agent -c conf/ -n a1 -f job/app_logs/kafka-flume-hdfs.conf -Dflume.root.logger=INFO,console

參數說明：

  --conf conf/ ：表示配置文件存儲在 conf/ 目錄，簡寫爲 -c
  --name a1 ：表示給 agent 起名爲 a1（要與配置文件一致），簡寫爲 -n
  --conf-file job/app_logs/kafka-flume-hdfs.conf ：flume 本次啓動讀取的配置文件是在 job/app_logs 文件夾下的 kafka-flume-hdfs.conf 文件，簡寫爲 -f
  -Dflume.root.logger==INFO,console ：-D 表示 flume 運行時動態修改 flume.root.logger 參數屬性值，並將控制檯日誌打印級別設置爲 INFO 級別。日誌級別包括:log、info、warn、error

5.3.5 測試

1）啓動客戶端日誌生成程序
2）查看 HDFS 的 /user/centos/applogs 路徑上是否有數據收到，以下圖所示：

六數據處理模塊框架搭建

6.1 數據處理模塊集羣部署規劃

6.2 配置 Hive 元數據存儲到 mysql

具體安裝配置，詳見：http://www.javashuo.com/article/p-baxyrovp-dy.html
使用 root 用戶，查看 mysql 狀態

[root@hadoop102 mysql-libs]# service mysql status

使用 root 用戶，啓動 mysql

[root@hadoop102 mysql-libs]# service mysql start

6.3 配置 Hive 支持 JSON 存儲

在 Hive 中採用 Json做爲存儲格式，須要建表時指定 Serde。insert into 時，Hive 使用 json 格式進行保存，查詢時，經過 json 庫進行解析。Hive 默認輸出是壓縮格式，這裏改爲不壓縮。
1）實操
（1）將json-serde-1.3.8-jar-with-dependencies.jar 導入到 hive 的 /opt/module/hive/lib 路徑下。
（2）在 /opt/module/hive/conf/hive-site.xml 文件中添加以下配置：

        <property>
            <name>hive.aux.jars.path</name>
            <value>file:///opt/module/hive/lib/json-serde-1.3.8-jar-with-dependencies.jar</value>
        </property>

        <!-- 輸出不壓縮 -->
        <property>
            <name>hive.exec.compress.output</name>
            <value>false</value>
        </property>

6.4 建立數據庫及分區表

0）啓動 hive

[atguigu@hadoop103 hive]$ bin/hive

查看數據庫
hive (default)> show databases;
若是 applogsdb 存在則刪除數據庫
hive (default)> drop database applogsdb;

1）建立數據庫

hive (default)> create database applogsdb;
使用 applogsdb 數據庫
hive (default)> use applogsdb;

2）建立分區表

--startup
CREATE external TABLE ext_startup_logs(
  createdAtMs bigint,
  appId string,
  tenantId string,
  deviceId string,
  appVersion string,
  appChannel string,
  appPlatform string,
  osType string,
  deviceStyle string,
  country string,
  province string,
  ipAddress string,
  network string,
  carrier string,
  brand string,
  screenSize string
)
PARTITIONED BY (ym string, day string,hm string) 
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe' 
STORED AS TEXTFILE;

--error
CREATE external TABLE ext_error_logs(
  createdAtMs bigint,
  appId string,
  tenantId string,
  deviceId string,
  appVersion string,
  appChannel string,
  appPlatform string,
  osType string,
  deviceStyle string,
  errorBrief string,
  errorDetail string
)
PARTITIONED BY (ym string, day string, hm string) 
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe' 
STORED AS TEXTFILE;

--event
CREATE external TABLE ext_event_logs(
  createdAtMs bigint,
  appId string,
  tenantId string,
  deviceId string,
  appVersion string,
  appChannel string,
  appPlatform string,
  osType string,
  deviceStyle string,
  eventId string,
  eventDurationSecs bigint,
  paramKeyValueMap Map<string, string>
)
PARTITIONED BY (ym string, day string, hm string) 
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe' 
STORED AS TEXTFILE;

--page
CREATE external TABLE ext_page_logs(
createdAtMs bigint,
  appId string,
  tenantId string,
  deviceId string,
  appVersion string,
  appChannel string,
  appPlatform string,
  osType string,
  deviceStyle string,
  pageViewCntInSession int,
  pageId string,
  visitIndex int,
  nextPage string,
  stayDurationSecs bigint
)
PARTITIONED BY (ym string, day string, hm string) 
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe' 
STORED AS TEXTFILE;

--usage
CREATE external TABLE ext_usage_logs(
  createdAtMs bigint,
  appId string,
  tenantId string,
  deviceId string,
  appVersion string,
  appChannel string,
  appPlatform string,
  osType string,
  deviceStyle string,
  singleUseDurationSecs bigint,
  singleUploadTraffic bigint,
  singleDownloadTraffic bigint
)
PARTITIONED BY (ym string, day string, hm string) 
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe' 
STORED AS TEXTFILE;

3）查看數據庫中的分區表

hive (applogsdb)> show tables;
tab_name
ext_error_logs
ext_event_logs
ext_page_logs
ext_startup_logs
ext_usage_logs

4）退出 hive

hive (applogsdb)> quit;

6.5 編寫 Hive 執行腳本

0）需求：實現每隔一分鐘將 HDFS 上的數據，導入到 Hive 對應分區中一次。
1）Date 命令
（1）明天

[atguigu@hadoop102 module]$ date -d "1 day" +%Y%m%d
20190527

（2）昨天

[atguigu@hadoop102 module]$ date -d "-1 day" +%Y%m%d
20190525

（3）上一個月

[atguigu@hadoop102 module]$ date -d "-1 month" +%Y%m%d
20190426

（4）前三分鐘

[atguigu@hadoop102 module]$ date -d "-3 minute" +%Y%m-%d-%H%M
201905-26-2133

（5）用「-」分割，截取出第一個參數

[atguigu@hadoop102 module]$ date -d "-3 minute" +%Y%m-%d-%H%M | awk -F '-' '{print $1}'
201905

（6）用「-」分割，截取出第二個參數

[atguigu@hadoop102 module]$ date -d "-3 minute" +%Y%m-%d-%H%M | awk -F '-' '{print $2}'
26

（7）用「-」分割，截取出第三個參數

[atguigu@hadoop102 module]$ date -d "-3 minute" +%Y%m-%d-%H%M | awk -F '-' '{print $3}'
2135

2）在 /home/atguigu/bin/applogs 目錄下編寫 shell 腳本（hdfstohive.sh）
（1）建立文件夾：

[atguigu@hadoop102 bin]$ mkdir applogs

（2）建立腳本 hdfstohive.sh

[atguigu@hadoop102 applogs]$ touch hdfstohive.sh

內容以下：

#!/bin/bash
systime=`date -d "-3 minute" +%Y%m-%d-%H%M`
ym=`echo ${systime} | awk -F '-' '{print $1}'`
day=`echo ${systime} | awk -F '-' '{print $2}'`
hm=`echo ${systime} | awk -F '-' '{print $3}'`

#執行 hive 的命令
hive -e "load data inpath '/user/centos/applogs/startup/${ym}/${day}/${hm}' into table applogsdb.ext_startup_logs partition(ym='${ym}',day='${day}',hm='${hm}')"
hive -e "load data inpath '/user/centos/applogs/error/${ym}/${day}/${hm}' into table applogsdb.ext_error_logs partition(ym='${ym}',day='${day}',hm='${hm}')"
hive -e "load data inpath '/user/centos/applogs/event/${ym}/${day}/${hm}' into table applogsdb.ext_event_logs partition(ym='${ym}',day='${day}',hm='${hm}')"
hive -e "load data inpath '/user/centos/applogs/usage/${ym}/${day}/${hm}' into table applogsdb.ext_usage_logs partition(ym='${ym}',day='${day}',hm='${hm}')"
hive -e "load data inpath '/user/centos/applogs/page/${ym}/${day}/${hm}' into table applogsdb.ext_page_logs partition(ym='${ym}',day='${day}',hm='${hm}')"

（3）修改腳本 hdfstohive.sh 權限

[atguigu@hadoop102 applogs]$ chmod 777 hdfstohive.sh

（4）必須保證 hive 的環境變量已經配置

[root@hadoop102 applogs]# vim /etc/profile
#HIVE_HOME
export HIVE_HOME=/opt/module/hive
export PATH=$PATH:$HIVE_HOME/bin

[root@hadoop102 applogs]# source /etc/profile

（5）執行一下 hive 腳本

[atguigu@hadoop102 applogs]$ ./hdfstohive.sh

（6）查看 hive 數據庫

[atguigu@hadoop102 hive]$ bin/hive
hive (default)> show databases;
OK
database_name
applogsdb
default
Time taken: 0.818 seconds, Fetched: 2 row(s)
hive (default)> use applogsdb;
OK
Time taken: 0.023 seconds
hive (applogsdb)> show tables;
OK
tab_name
ext_error_logs
ext_event_logs
ext_page_logs
ext_startup_logs
ext_usage_logs
Time taken: 0.022 seconds, Fetched: 5 row(s)
hive (applogsdb)> select * from ext_error_logs;

6.6 編寫 Linux 調度 crondtab

1）crontab 經常使用命令
（1）查看狀態：

service crond status

（2）中止狀態：

service crond stop

（3）啓動狀態：

service crond start

（4）編輯 crontab 定時任務

crontab -e

（5）查詢 crontab 任務

crontab -l

（6）刪除當前用戶全部的 crontab 任務

crontab -r

2）編寫 crontab 調度
（1）進入編寫 crontab 調度

[atguigu@hadoop102 applogs]$ crontab -e

（2）實現每分鐘執行一次

* * * * * source /etc/profile;/home/atguigu/bin/applogs/hdfstohive.sh

參數解釋：

（3）查看 crontab

[atguigu@hadoop102 applogs]$ crontab -l
* * * * * source /etc/profile;/home/atguigu/bin/applogs/hdfstohive.sh

6.7 測試

1）啓動客戶端日誌生成程序
2）等待三分鐘後，檢查 HDFS 的 /user/hive/warehouse/applogsdb.db 路徑是否有新數據產生。
3）查詢全部啓動日誌信息

[atguigu@hadoop102 hive]$ bin/hive
hive (default)> use applogsdb;
hive (applogsdb)> select * from ext_startup_logs;

4）查詢指定 app 的用戶數

hive (applogsdb)> select count(distinct deviceid) from ext_startup_logs where appid='sdk34734';

5）統計新增用戶
須要自定義 UDF 函數。

七業務需求處理

7.1 自定義 UDF 函數

0）需求：
根據輸入的時間信息，返回當天的起始時間；
根據輸入的時間信息，返回本週的起始時間；
根據輸入的時間信息，返回本月的起始時間；
根據輸入的時間和時間格式化信息，返回按照格式化要求顯示的信息。

1）在 jar 工程：app_logs_hive 中的 com.atguigu.hive 目錄下編寫獲取日期開始時間、周開始時間、和月開始時間的工具類

package com.atguigu.hive;

import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;

public class DateUtil {
    /**
     * 獲得指定 date 所在天的的零時刻
     */
    public static Date getDayBeginTime(Date d) {
        try {
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy/MM/dd 00:00:00");

            return sdf.parse(sdf.format(d));
        } catch (Exception e) {
            e.printStackTrace();
        }

        return null;
    }

    /**
     * 獲得指定 date 所在天的零時刻（+偏移量）
     */
    public static Date getDayBeginTime(Date d, int offset) {
        try {
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy/MM/dd 00:00:00");
            Date beginDate = sdf.parse(sdf.format(d));
            Calendar c = Calendar.getInstance();
            c.setTime(beginDate);
            c.add(Calendar.DAY_OF_MONTH, offset);

            return c.getTime();
        } catch (Exception e) {
            e.printStackTrace();
        }

        return null;
    }

    /**
     * 獲得指定 date 所在周的起始時刻
     */
    public static Date getWeekBeginTime(Date d) {
        try {
            // 獲得 d 的所在天的的零時刻
            Date beginDate = getDayBeginTime(d);
            Calendar c = Calendar.getInstance();
            c.setTime(beginDate);
            int n = c.get(Calendar.DAY_OF_WEEK);

            c.add(Calendar.DAY_OF_MONTH, -(n - 1));

            return c.getTime();
        } catch (Exception e) {
            e.printStackTrace();
        }

        return null;
    }

    /**
     * 獲得指定 date 所在周的起始時刻（+偏移量）
     */
    public static Date getWeekBeginTime(Date d, int offset) {
        try {
            // 獲得 d 的所在天的的零時刻
            Date beginDate = getDayBeginTime(d);
            Calendar c = Calendar.getInstance();
            c.setTime(beginDate);
            int n = c.get(Calendar.DAY_OF_WEEK);

            // 定位到本週第一天
            c.add(Calendar.DAY_OF_MONTH, -(n - 1));
            c.add(Calendar.DAY_OF_MONTH, offset * 7);

            return c.getTime();
        } catch (Exception e) {
            e.printStackTrace();
        }

        return null;
    }

    /**
     * 獲得指定 date 所在月的起始時刻
     */
    public static Date getMonthBeginTime(Date d) {
        try {
            // 獲得 d 的所在天的的零時刻
            Date beginDate = getDayBeginTime(d);
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy/MM/01 00:00:00");

            return sdf.parse(sdf.format(beginDate));
        } catch (Exception e) {
            e.printStackTrace();
        }

        return null;
    }

    /**
     * 獲得指定 date 所在月的偏移時刻（+偏移量）
     */
    public static Date getMonthBeginTime(Date d, int offset) {
        try {
            // 獲得 d 的所在天的的零時刻
            Date beginDate = getDayBeginTime(d);
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy/MM/01 00:00:00");

            // 獲得 d 所在月的第一天的零時刻
            Date firstDay = sdf.parse(sdf.format(beginDate));

            Calendar c = Calendar.getInstance();
            c.setTime(firstDay);

            // 對月進行滾動
            c.add(Calendar.MONTH, offset);

            return c.getTime();
        } catch (Exception e) {
            e.printStackTrace();
        }

        return null;
    }
}

2）編寫 DayBeginUDF、WeekBeginUDF、MonthBeginUDF、FormatTimeUDF 函數
（1）建立 java 工程導入 pom.xml 文件
建立 jar 工程：app_logs_hive
添加 maven 框架支持，並導入 pom 文件，並刷新一下 maven

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.atguigu</groupId>
    <artifactId>app_logs_hive</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>

        <dependency>
            <groupId>org.apache.hive</groupId>
            <artifactId>hive-exec</artifactId>
            <version>1.2.1</version>
        </dependency>
    </dependencies>
</project>

（2）編寫 DayBeginUDF
建立包名：com.atguigu.hive
編寫代碼：

package com.atguigu.hive;

import org.apache.hadoop.hive.ql.exec.UDF;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * 計算 day 的起始毫秒數
 */
public class DayBeginUDF extends UDF {

    // 計算如今的起始時刻(毫秒數)
    public long evaluate() throws ParseException {
        return evaluate(new Date());
    }

    // 指定天偏移量
    public long evaluate(int offset) throws ParseException {
        return evaluate(DateUtil.getDayBeginTime(new Date(), offset));
    }

    // 計算某天的起始時刻，日期類型(毫秒數)
    public long evaluate(Date d) throws ParseException {
        return DateUtil.getDayBeginTime(d).getTime();
    }

    // 計算某天的起始時刻，日期類型，帶偏移量(毫秒數)
    public long evaluate(Date d, int offset) throws ParseException {
        return DateUtil.getDayBeginTime(d, offset).getTime();
    }

    // 計算某天的起始時刻，String 類型(毫秒數)
    public long evaluate(String dateStr) throws ParseException {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
        Date d = sdf.parse(dateStr);

        return evaluate(d);
    }

    // 計算某天的起始時刻，String 類型，帶偏移量(毫秒數)
    public long evaluate(String dateStr, int offset) throws ParseException {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
        Date d = sdf.parse(dateStr);

        return DateUtil.getDayBeginTime(d, offset).getTime();
    }

    // 計算某天的起始時刻，String 類型，帶格式化要求(毫秒數)
    public long evaluate(String dateStr, String fmt) throws ParseException {
        SimpleDateFormat sdf = new SimpleDateFormat(fmt);
        Date d = sdf.parse(dateStr);

        return DateUtil.getDayBeginTime(d).getTime();
    }

    // 計算某天的起始時刻，String 類型，帶格式化，帶偏移量(毫秒數)
    public long evaluate(String dateStr, String fmt, int offset) throws ParseException {
        SimpleDateFormat sdf = new SimpleDateFormat(fmt);
        Date d = sdf.parse(dateStr);

        return DateUtil.getDayBeginTime(d, offset).getTime();
    }
}

（3）編寫 WeekBeginUDF

package com.atguigu.hive;

import org.apache.hadoop.hive.ql.exec.UDF;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * 計算 week 的起始毫秒數
 */
public class WeekBeginUDF extends UDF {

    // 計算本週的起始時間，(毫秒數)
    public long evaluate() throws ParseException {
        return DateUtil.getWeekBeginTime(new Date()).getTime();
    }

    // 指定周偏移量
    public long evaluate(int offset) throws ParseException {
        return DateUtil.getWeekBeginTime(new Date(), offset).getTime();
    }

    // 計算某周的起始時刻，日期類型(毫秒數)
    public long evaluate(Date d) throws ParseException {
        return DateUtil.getWeekBeginTime(d).getTime();
    }

    // 計算某周的起始時刻，日期類型，帶偏移量(毫秒數)
    public long evaluate(Date d, int offset) throws ParseException {
        return DateUtil.getWeekBeginTime(d, offset).getTime();
    }

    // 計算某周的起始時刻，String 類型(毫秒數)
    public long evaluate(String dateStr) throws ParseException {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
        Date d = sdf.parse(dateStr);

        return DateUtil.getWeekBeginTime(d).getTime();
    }

    // 計算某周的起始時刻，String 類型，帶偏移量(毫秒數)
    public long evaluate(String dateStr, int offset) throws ParseException {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
        Date d = sdf.parse(dateStr);
        return DateUtil.getWeekBeginTime(d, offset).getTime();
    }

    // 計算某周的起始時刻，String 類型，帶格式化要求(毫秒數)
    public long evaluate(String dateStr, String fmt) throws ParseException {
        SimpleDateFormat sdf = new SimpleDateFormat(fmt);
        Date d = sdf.parse(dateStr);

        return DateUtil.getWeekBeginTime(d).getTime();
    }

    // 計算某周的起始時刻，String 類型，帶格式化，帶偏移量(毫秒數)
    public long evaluate(String dateStr, String fmt, int offset) throws ParseException {
        SimpleDateFormat sdf = new SimpleDateFormat(fmt);
        Date d = sdf.parse(dateStr);

        return DateUtil.getWeekBeginTime(d, offset).getTime();
    }
}

（4）編寫 MonthBeginUDF

package com.atguigu.hive;

import org.apache.hadoop.hive.ql.exec.UDF;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * 計算 month 的起始毫秒數
 */
public class MonthBeginUDF extends UDF {

    // 計算本月的起始時刻(毫秒數)
    public long evaluate() throws ParseException {
        return DateUtil.getMonthBeginTime(new Date()).getTime();
    }

    // 指定月偏移量
    public long evaluate(int offset) throws ParseException {
        return DateUtil.getMonthBeginTime(new Date(), offset).getTime();
    }

    // 計算某月的起始時刻，日期類型(毫秒數)
    public long evaluate(Date d) throws ParseException {
        return DateUtil.getMonthBeginTime(d).getTime();
    }

    // 計算某月的起始時刻，日期類型，帶偏移量(毫秒數)
    public long evaluate(Date d, int offset) throws ParseException {
        return DateUtil.getMonthBeginTime(d, offset).getTime();
    }

    // 計算某月的起始時刻，String 類型(毫秒數)
    public long evaluate(String dateStr) throws ParseException {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
        Date d = sdf.parse(dateStr);

        return DateUtil.getMonthBeginTime(d).getTime();
    }

    // 計算某月的起始時刻，String 類型，帶偏移量(毫秒數)
    public long evaluate(String dateStr, int offset) throws ParseException {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
        Date d = sdf.parse(dateStr);

        return DateUtil.getMonthBeginTime(d, offset).getTime();
    }

    // 計算某月的起始時刻，String 類型，帶格式化要求(毫秒數)
    public long evaluate(String dateStr, String fmt) throws ParseException {
        SimpleDateFormat sdf = new SimpleDateFormat(fmt);
        Date d = sdf.parse(dateStr);

        return DateUtil.getMonthBeginTime(d).getTime();
    }

    // 計算某月的起始時刻，String 類型，帶格式化，帶偏移量(毫秒數)
    public long evaluate(String dateStr, String fmt, int offset) throws ParseException {
        SimpleDateFormat sdf = new SimpleDateFormat(fmt);
        Date d = sdf.parse(dateStr);

        return DateUtil.getMonthBeginTime(d, offset).getTime();
    }
}

（5）編寫 FormatTimeUDF

package com.atguigu.hive;

import org.apache.hadoop.hive.ql.exec.UDF;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * 格式化時間
 */
public class FormatTimeUDF extends UDF {

    // 根據輸入的時間毫秒值（long 類型）和格式化要求，返回 String 類型時間
    public String evaluate(long ms, String fmt) throws ParseException {

        SimpleDateFormat sdf = new SimpleDateFormat(fmt);
        Date d = new Date();
        d.setTime(ms);

        return sdf.format(d);
    }

    // 根據輸入的時間毫秒值（String 類型）和格式化要求，返回 String 類型時間
    public String evaluate(String ms, String fmt) throws ParseException {
        SimpleDateFormat sdf = new SimpleDateFormat(fmt);
        Date d = new Date();
        d.setTime(Long.parseLong(ms));

        return sdf.format(d);
    }

    // 根據輸入的時間毫秒值（long 類型）、格式化要求，和區分周的任意值，返回 String 類型時間
    public String evaluate(long ms, String fmt, int week) throws ParseException {
        Date d = new Date();
        d.setTime(ms);

        // 周內第一天
        Date firstDay = DateUtil.getWeekBeginTime(d);
        SimpleDateFormat sdf = new SimpleDateFormat(fmt);

        return sdf.format(firstDay);
    }
}

3）導出 jar 包（app_logs_hive.jar）
4）添加 app_logs_hive.jar 到類路徑 /opt/module/hive/lib 下
（1）臨時添加 jar 包：（每次關閉 hive 後，再次鏈接 hive 時須要再次添加，太麻煩！）

hive (applogsdb)> add jar /opt/module/hive/lib/app_logs_hive.jar;

（2）永久添加 jar 包：
在 hive-site.xml 文件中添加：

        <property>
            <name>hive.aux.jars.path</name>
            <value>file:///opt/module/hive/lib/app_logs_hive.jar</value>
        </property>

因爲以前添加過 json 的 jar 包，因此修改成以下方式：

        <!-- 能夠添加支持 json 格式的 jar 包，自定義的 UDF jar 包等等 -->
        <property>
            <name>hive.aux.jars.path</name>
            <value>file:///opt/module/hive/lib/json-serde-1.3.8-jar-with-dependencies.jar,file:///opt/module/hive/lib/app_logs_hive.jar</value>
        </property>

5）註冊爲永久函數（學習測試階段爲了學習的方便，註冊爲永久的函數）

hive (applogsdb)> create function getdaybegin AS 'com.atguigu.hive.DayBeginUDF';  # getdaybegin 是函數名稱，自定義的，起什麼均可以
hive (applogsdb)> create function getweekbegin AS 'com.atguigu.hive.WeekBeginUDF';
hive (applogsdb)> create function getmonthbegin AS 'com.atguigu.hive.MonthBeginUDF';
hive (applogsdb)> create function formattime AS 'com.atguigu.hive.FormatTimeUDF';

註冊爲臨時函數
hive (applogsdb)> create temporary function getdaybegin AS 'com.atguigu.hive.DayBeginUDF';

6）驗證函數
登陸 mysql

[atguigu@hadoop102 ~]$ mysql -uroot -p123456
mysql> show databases;
mysql> use metastore;
mysql> show tables;
mysql> select * from FUNCS; # 查看註冊的永久函數

查看結果截圖

7）刪除函數

hive (applogsdb)> drop function getdaybegin;
hive (applogsdb)> drop function getweekbegin;
hive (applogsdb)> drop function getmonthbegin;
hive (applogsdb)> drop function formattime;

8）注意：在哪一個數據庫中註冊的永久函數，必須在哪一個數據庫下將該方法刪除！！！
好比：在 applogsdb 數據庫中建立的函數，必須在該數據中調用 drop 方法才能實現刪除功能。

7.2 新增用戶統計

7.2.1 任意日新增用戶

1）今天新增用戶
（1）判斷今天新增用戶條件：
先按照設備 id 分組；
再根據建立該日誌的最開始時間，是否在今天範圍內。
（2）統計今天新增個數

select
  count(*)
from
  (select min(createdatms) as mintime
    from ext_startup_logs
    where appid = 'sdk34734'
    group by deviceid
    having mintime >= getdaybegin() and mintime < getdaybegin(1)
  ) t;

2）昨天新增用戶

select
  count(*)
from
  (select min(createdatms) as mintime
    from ext_startup_logs
    where appid = 'sdk34734'
    group by deviceid
    having mintime >= getdaybegin(-1) and mintime < getdaybegin()
  ) t;

3）指定時間的新增用戶

select
  count(*)
from
  (select min(createdatms) as mintime
    from ext_startup_logs
    where appid = 'sdk34734'
    group by deviceid
    having mintime >= getdaybegin('2019/5/26 00:00:00') and mintime < getdaybegin('2019/05/26 00:00:00',1)
  ) t;

7.2.2 任意周新增用戶

1）本週新增用戶

select
  count(*)
from
  (select min(createdatms) as mintime
    from ext_startup_logs
    where appid = 'sdk34734'
    group by deviceid
    having mintime >= getweekbegin() and mintime < getweekbegin(1)
  ) t;

2）上一週新增用戶

select
  count(*)
from
  (select min(createdatms) as mintime
    from ext_startup_logs
    where appid = 'sdk34734'
    group by deviceid
    having mintime >= getweekbegin(-1) and mintime < getweekbegin()
  ) t;

3）指定周時間的新增用戶

select
  count(*)
from
  (select min(createdatms) as mintime
    from ext_startup_logs
    where appid = 'sdk34734'
    group by deviceid
    having mintime >= getweekbegin('2019/05/26 00:00:00') and mintime < getweekbegin('2019/05/26 00:00:00',1)
  ) t;

7.2.3 月新增用戶

select
  count(*)
from
  (select min(createdatms) as mintime
    from ext_startup_logs
    where appid = 'sdk34734'
    group by deviceid
    having mintime >= getmonthbegin() and mintime < getmonthbegin(1)
  ) t;

7.3 活躍用戶統計

7.3.1 日、周、月活躍用戶

0）分析：

1）日活躍用戶數

select
  count(distinct deviceid)
from ext_startup_logs
where 
  appid = 'sdk34734' and createdatms >= getdaybegin() and createdatms < getdaybegin(1);

2）周活躍用戶數

select
  count(distinct deviceid)
from ext_startup_logs
where 
  appid = 'sdk34734' and createdatms >= getweekbegin() and createdatms < getweekbegin(1);

3）月活躍用戶數

select
  count(distinct deviceid)
from ext_startup_logs
where 
  appid = 'sdk34734' and createdatms >= getmonthbegin() and createdatms < getmonthbegin(1);

7.3.2 指定時間內查詢日活、周活、月活

0）分析：

1）一次查詢出一週內，天天的日活躍數

select
  formattime(createdatms,'yyyy/MM/dd') as day, count(distinct deviceid) as count
from ext_startup_logs
where appid = 'sdk34734' and createdatms >= getweekbegin() and createdatms < getweekbegin(1)
group by formattime(createdatms,'yyyy/MM/dd');

2）一次查詢出過去的 5 周，每週的周活躍數

select
  formattime(createdatms,'yyyy/MM/dd',0) as week ,count(distinct deviceid) as count
from ext_startup_logs
where appid = 'sdk34734' and createdatms >= getweekbegin(-6) and createdatms < getweekbegin(-1)
group by formattime(createdatms,'yyyy/MM/dd',0);

3）一次查詢出過去的三個月內，每週的月活躍數

select
  formattime(createdatms,'yyyy/MM',0) as month ,count(distinct deviceid) as count
from ext_startup_logs
where appid = 'sdk34734' and createdatms >= getmonthbegin(-4) and createdatms < getmonthbegin(-1)
group by formattime(createdatms,'yyyy/MM',0);

7.3.3 優化活躍數查詢

根據時間分區表去查詢，避免全表掃描

select
  count(distinct deviceid)
from ext_startup_logs
where appid = 'sdk34734' and ym = formattime(getdaybegin(),'yyyyMM') and day = formattime(getdaybegin(),'dd');

7.3.4 過去五週周活躍用戶數

過去的五週(包含本週)某個 app 每週的周活躍用戶數
鏈接函數測試：select concat(ym,day) from ext_startup_logs;

select
  formattime(createdatms,'yyyyMMdd',0) as stdate, count(distinct deviceid) as stcount
from ext_startup_logs
where concat(ym,day) >= formattime(getweekbegin(-4),'yyyyMMdd') and appid ='sdk34734'
group by formattime(createdatms,'yyyyMMdd',0);

7.3.5 過去六月活躍用戶數

最近的六個月(包含本月)每個月的月活躍數。

select
  formattime(createdatms,'yyyyMM') as stdate, count(distinct deviceid) as stcount
from ext_startup_logs
where ym >= formattime(getmonthbegin(-5),'yyyyMM') and appid ='sdk34734'
group by formattime(createdatms,'yyyyMM');

7.3.6 連續 n 周活躍用戶統計

分析：

連續活躍3周

select 
  deviceid, count(distinct(formattime(createdatms,'yyyyMMdd',0))) c
from ext_startup_logs
where appid = 'sdk34734' and concat(ym,day) >= formattime(getweekbegin(-2),'yyyyMMdd')
group by deviceid
having c = 3;

7.3.7 忠誠用戶

忠誠用戶（連續活躍5周）

select 
  deviceid, count(distinct(formattime(createdatms,'yyyyMMdd',0))) c
from ext_startup_logs
where appid = 'sdk34734' and concat(ym,day) >= formattime(getweekbegin(-4),'yyyyMMdd')
group by deviceid
having c = 5;

7.4 沉默用戶統計

分析：

查詢沉默用戶數（一共只有一條日誌；且安裝時間超過2天）

select
  count(*)
from
  (select deviceid, count(createdatms) as dcount, min(createdatms) as dmin
    from ext_startup_logs
    where appid = 'sdk34734'
    group by deviceid
    having dcount = 1 and dmin < getdaybegin(-1)
  ) t;

7.5 啓動次數統計

今天 app 的啓動次數：啓動次數相似於活躍用戶數，活躍用戶數去重，啓動次數不須要去重。

select
  count(deviceid)
from ext_startup_logs
where appid = 'sdk34734' and ym = formattime(getdaybegin(),'yyyyMM') and day = formattime(getdaybegin(),'dd');

7.6 版本分佈統計

1）今天 appid 爲 34734 的不一樣版本的活躍用戶數。

select
  appversion, count(distinct deviceid)
from ext_startup_logs
where appid = 'sdk34734' and ym = formattime(getdaybegin(),'yyyyMM') and day = formattime(getdaybegin(),'dd')
group by appversion;

2）本週內天天各版本日活躍數
（1）本週內：where concat(ym,day) >= formattime(getweekbegin(),'yyyyMMdd')
（2）天天：group by formattime(createdatms,'yyyyMMdd')
（2）各個版本：group by appversion
（3）日活躍數：count(distinct deviceid)

select
  formattime(createdatms,'yyyyMMdd'), appversion, count(distinct deviceid)
from ext_startup_logs
where appid = 'sdk34734' and concat(ym,day) >= formattime(getweekbegin(),'yyyyMMdd')
group by formattime(createdatms,'yyyyMMdd'), appversion;

7.7 留存分析統計

7.7.1 本週迴流用戶統計

本週迴流用戶：上週沒有啓動過，本週啓動過。
分析：

sql 語句以下：

select
  distinct s.deviceid
from ext_startup_logs s
where appid = 'sdk34734' and concat(ym,day) >= formattime(getweekbegin(),'yyyyMMdd') and deviceid not in 
  (select
    distinct t.deviceid
  from ext_startup_logs t
  where t.appid = 'sdk34734' and concat(t.ym,t.day) >= formattime(getweekbegin(-1),'yyyyMMdd') and concat(t.ym,t.day) < formattime(getweekbegin(),'yyyyMMdd')
);

7.7.2 連續n周沒有啓動的用戶

連續 2 周內沒有啓動過：本週和上週沒啓動過；大上週啓動過。
sql 語句以下：

select
  distinct s.deviceid
from ext_startup_logs s
where appid='sdk34734'
and concat(ym,day) >= formattime(getweekbegin(-2),'yyyyMMdd')
and concat(ym,day) < formattime(getweekbegin(-1),'yyyyMMdd')
and deviceid not in 
  (select
    distinct(t.deviceid)
  from ext_startup_logs t
  where t.appid='sdk34734'
  and concat(t.ym,t.day) >= formattime(getweekbegin(-1),'yyyyMMdd')
);

7.7.3 留存用戶統計

本週留存用戶=上週新增用戶和本週活躍用戶的交集。
分析：

sql 語句以下：

select
  distinct s.deviceid
from ext_startup_logs s
where appid = 'sdk34734'
and concat(ym,day) >= formattime(getweekbegin(-1),'yyyyMMdd')
and concat(ym,day) < formattime(getweekbegin(),'yyyyMMdd')
and deviceid in (
select distinct t.deviceid
from (
  select tt.deviceid , min(tt.createdatms) mintime
  from ext_startup_logs tt
  where tt.appid = 'sdk34734'
  group by tt.deviceid having mintime >= getweekbegin(-2) and mintime < getweekbegin(-1)
) t);

7.8 新鮮度分析

用戶新鮮度 = 某段時間的新增用戶數/某段時間的活躍的用戶數 .
1）今天新增用戶（爲n）

select
  count(*)
from
  (select min(createdatms) mintime
  from ext_startup_logs
  where appid = 'sdk34734'
  group by deviceid
  having mintime >= getdaybegin() and mintime < getdaybegin(1)
  ) t;

2）今天活躍用戶（m）

select
  count(distinct deviceid)
from ext_startup_logs
where appid = 'sdk34734' and createdatms >= getdaybegin() and createdatms < getdaybegin(1);

3）新鮮度 = n / m
注意判斷 m 等於0的狀況。

八數據展現模塊

8.1 建立 web 可視化工程

8.1.1 建立 web 工程導入 pom.xml

1）建立一個 web 工程：app_logs_visualize_web
2）添加maven框架支持
3）導入pom文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.atguigu</groupId>
    <artifactId>app_logs_visualize_web</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>war</packaging>

    <!-- 日誌框架版本號 -->
    <properties>
        <log4j.version>1.2.17</log4j.version>
        <slf4j.version>1.7.22</slf4j.version>
    </properties>

    <dependencies>
        <!-- 日誌框架 -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>jcl-over-slf4j</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>${log4j.version}</version>
        </dependency>

        <!-- 單元測試 -->
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.11</version>
        </dependency>
        <!-- mybatis 框架 -->
        <dependency>
            <groupId>org.mybatis</groupId>
            <artifactId>mybatis</artifactId>
            <version>3.2.1</version>
        </dependency>
        <dependency>
            <groupId>c3p0</groupId>
            <artifactId>c3p0</artifactId>
            <version>0.9.1.2</version>
        </dependency>
        <dependency>
            <groupId>org.mybatis</groupId>
            <artifactId>mybatis-spring</artifactId>
            <version>1.3.0</version>
        </dependency>

        <!-- spring 框架 -->
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-webmvc</artifactId>
            <version>4.3.3.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-jdbc</artifactId>
            <version>4.3.3.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>org.aspectj</groupId>
            <artifactId>aspectjweaver</artifactId>
            <version>1.8.10</version>
        </dependency>
        <dependency>
            <groupId>javax.servlet</groupId>
            <artifactId>javax.servlet-api</artifactId>
            <version>3.0.1</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>jstl</groupId>
            <artifactId>jstl</artifactId>
            <version>1.2</version>
        </dependency>
        <dependency>
            <groupId>taglibs</groupId>
            <artifactId>standard</artifactId>
            <version>1.1.2</version>
        </dependency>

        <!--jackson json 解析框架-->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-core</artifactId>
            <version>2.8.8</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.8.3</version>
        </dependency>

        <!-- hive 框架 -->
        <dependency>
            <groupId>org.apache.hive</groupId>
            <artifactId>hive-jdbc</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.eclipse.jetty.aggregate</groupId>
                    <artifactId>jetty-all</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.mortbay.jetty</groupId>
                    <artifactId>jetty</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>tomcat</groupId>
                    <artifactId>jasper-compiler</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>tomcat</groupId>
                    <artifactId>jasper-runtime</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.mortbay.jetty</groupId>
                    <artifactId>jsp-2.1</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.mortbay.jetty</groupId>
                    <artifactId>jsp-api-2.1</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.eclipse.jetty.orbit</groupId>
                    <artifactId>javax.servlet</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>javax.servlet</groupId>
                    <artifactId>servlet-api</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>javax.servlet</groupId>
                    <artifactId>jsp-api</artifactId>
                </