1.概述

　　本课程的视频教程地址：《用户行为分析之编码实践》

　　本课程以用户行为分析案例为基础，带着大家去完成对各个KPI的编码工作，以及应用调度工作，让大家通过本课程掌握Hadoop项目的编码、调度流程。下面我们来看看本课程有哪些课时，如下图所示：

　　首先，我们来学习第一课时：《Hadoop项目基础代码》。

2.内容

2.1 Hadoop项目基础代码

　　本课时介绍编写Hadoop基础代码及脚本，在前面搭建好的Hadoop项目工程上， 完成项目的基本代码的编写，以及一些注意事项，为编写核心代码做准备，让大家掌握Hadoop项目的基础代码开发。

　　本课时主要包含以下知识点，如下图所示：

　　下面，我为大家介绍Hadoop项目工程的基本信息配置，由于本课程属于编码实践，所以设计到编码的相关流程。 所以大家在学习的时候，请事先将集群启动，IDE打开。下面我给大家演示如下内容，他们分别是：

• 项目工程的相关配置信息（pom文件的依赖配置，log日志信息的配置）
• 集群信息的相关配置（连接集群节点路径信息） 

　　基础代码实现包含的内容，如下图所示：

　　具体演示细节，大家可以观看视频，这里我就不多做赘述了。《观看地址》

2.2 Hadoop项目核心地址实现

　　本课时介绍如何去实现Hadoop的核心代码模块， 在基础代码模块上，完成核心代码的实现，让大家掌握项目相关指标的统计开发。

　　下面我们来看看本课时有那些知识点，如下图所示：

　　下面我们来看看离线结果统计的处理方式有哪些，这里，我用一个图来说明，在离线统计中的统计方式，如下图所示：

　　这里，从图中我们可以看出，我们可以使用编写Hive脚本或Hive应用程序来统计， 也可以编写MapReduce程序来完成统计，也可以组合使用，这里，本课程的案例， 我使用的是组合使用，用Hive和MapReduce组合来完成。

　　接着来看核心代码实现的内容，如下图所示：

　　脚本如下所示：

#创建分区
CREATE EXTERNAL TABLE ubas(ip string, timespan string, url string,hour string)PARTITIONED BY (logdate string) 
ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LOCATION '/home/hdfs/ubas/out/meta'

　　统计的KPI脚本，如下所示：

# clean hdfs data and output 
/home/hadoop/hadoop-2.6.0/bin/hadoop jar ubas-1.0.0-jar-with-dependencies.jar $1

# use hive to stats

## 1.location data to partition
/home/hadoop/hive-0.14.0-bin/bin/hive -e "ALTER TABLE ubas ADD PARTITION(logdate='$1') LOCATION '/home/hdfs/ubas/out/meta/$1';"

## 2.stats pv
/home/hadoop/hive-0.14.0-bin/bin/hive -e "CREATE TABLE pv_$1 AS SELECT COUNT(1) AS PV FROM ubas WHERE logdate='$1';"

## 3.stats ip
/home/hadoop/hive-0.14.0-bin/bin/hive -e "CREATE TABLE ip_$1 AS SELECT COUNT(DISTINCT ip) AS IP FROM ubas WHERE logdate='$1';"

## 4.stats amount hour
/home/hadoop/hive-0.14.0-bin/bin/hive -e "CREATE TABLE amount_$1 ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' AS SELECT '$1',hour AS HOUR_TAG, COUNT(hour) AS HOUR,'' AS UPDATE_DATE FROM ubas WHERE logdate='$1' GROUP BY hour;"

## 5.stats jr
/home/hadoop/hive-0.14.0-bin/bin/hive -e "CREATE TABLE jr_$1 AS SELECT COUNT(1) AS JR FROM (SELECT COUNT(ip) AS times FROM ubas WHERE logdate='$1' GROUP BY ip HAVING times=1) e;"

## 6.combine pv,ip,jr and tr to ubas table
/home/hadoop/hive-0.14.0-bin/bin/hive -e "CREATE TABLE ubas_$1 ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' AS SELECT '$1', a.pv, b.ip, c.jr, ROUND(COALESCE(CAST(b.ip AS DOUBLE), 0)/a.pv, 2),'' AS UPDATE_DATE FROM pv_$1 a JOIN ip_$1 b ON 1=1 JOIN jr_$1 c ON 1=1 ;"

# sqoop data to mysql

## 1.sqoop t_kpi_day
/home/hadoop/sqoop-1.4.5/bin/sqoop export -D sqoop.export.records.per.statement=100 --connect jdbc:mysql://10.211.55.26:3306/hadoop_ubas --username root --password root --table t_kpi_day --fields-terminated-by ',' --export-dir "/home/hive/warehouse/ubas_$1" --batch --update-key createdate --update-mode allowinsert;

## 2.sqoop t_kpi_hour
/home/hadoop/sqoop-1.4.5/bin/sqoop export -D sqoop.export.records.per.statement=100 --connect jdbc:mysql://10.211.55.26:3306/hadoop_ubas --username root --password root --table t_kpi_hour --fields-terminated-by ',' --export-dir "/home/hive/warehouse/amount_$1" --batch --update-key createdate,kpi_code --update-mode allowinsert;

# drop tmp table to hive warehouse
/home/hadoop/hive-0.14.0-bin/bin/hive -e "drop table amount_$1;drop table ip_$1;drop table jr_$1;drop table pv_$1;drop table ubas_$1;"

2.3 统计结果处理

　　本课时介绍将统计好的数据导出到关系型数据库，以及对外提供数据共享接口，让大家掌握导出数据的流程及共享接口程序的编写。

 　　本课时主要有一下知识点，如下图所示：

　　下面我们来看看使用 Sqoop 如何将 HDFS 上的统计结果导出到 MySQL 数据库， 接下来，我们来看看 Sqoop 的导出流程，如下图所示：

　　首先，我们是将统计结果存放在 HDFS 集群上，然后我们使用 Sqoop 工具去将 HDFS 的数据导出到关系型数据库，如 MySQL 整个基本流程就是这样。下面我们来使用 Sqoop 工具对HDFS 上的数据进行导出，同样，在使用导出功能时，这样大家需要 安装 Sqoop 工具，Sqoop 的安装较为简单，大家可以下去补充学习下，这里就不多做赘述了。

　　接下来，我们来看看数据共享流程，如下图所示：

　　从图中我们可以看出，我们将统计后的结果存放在mysql数据库中，这时我们需要编写一个rpc将数据共享出去，这里我采用的共享方式是， 编写一个thrift的服务接口，将数据通过这个接口共享出去，然后，前端同学获取数据后，可以将数据结果以图表的方式进行展示。

　　Thrift接口代码，如下所示：

• Thrift接口文件

namespace java cn.jikexueyuan.ubas.service

service UBASService {
    map<string, double> queryDayKPI(1:string beginDate,2:string endDate),
    map<double, double> queryHourKPI(1:string beginDate,2:string endDate)
}

package cn.jikexueyuan.ubas.main;

import org.apache.thrift.TProcessorFactory;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.server.THsHaServer;
import org.apache.thrift.server.TServer;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.jikexueyuan.ubas.service.UBASService;
import cn.jikexueyuan.ubas.service.impl.UBASServiceImpl;

/**
 * @Date Mar 23, 2015
 *
 * @Author dengjie
 */
public class StatsServer {

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

    private final int PORT = 9090;

    @SuppressWarnings({ "rawtypes", "unchecked" })
    private void start() {
        try {
            TNonblockingServerSocket socket = new TNonblockingServerSocket(PORT);
            final UBASService.Processor processor = new UBASService.Processor(new UBASServiceImpl());
            THsHaServer.Args arg = new THsHaServer.Args(socket);
            /*
             * Binary coded format efficient, intensive data transmission, The
             * use of non blocking mode of transmission, according to the size
             * of the block, similar to the Java of NIO
             */
            arg.protocolFactory(new TCompactProtocol.Factory());
            arg.transportFactory(new TFramedTransport.Factory());
            arg.processorFactory(new TProcessorFactory(processor));
            TServer server = new THsHaServer(arg);
            server.serve();
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    public static void main(String[] args) {
        try {
            logger.info("start thrift server...");
            StatsServer stats = new StatsServer();
            stats.start();
        } catch (Exception ex) {
            ex.printStackTrace();
            logger.error(String.format("run thrift server has error,msg is %s", ex.getMessage()));
        }
    }

}