PHP+Hadoop+Hive+Thrift+Mysql實現數據統計分析

原址：http://www.cnblogs.com/wicub/p/6094045.html

 

安裝

Hadoop安裝: http://www.powerxing.com/install-hadoop/ 
Hadoop集羣配置: http://www.powerxing.com/install-hadoop-cluster/ 
Hive安裝： https://chu888chu888.gitbooks.io/hadoopstudy/content/Content/8/chapter0807.html

安裝具體教程請看上面連接，本地測試只用了單機配置，集羣配置(後面的flume用到)看上面的詳細連接, 由於以前沒有接觸過java的相關，這裏說下遇到的幾個問題.

Hadoop和Hive的1.x和2.x版本要對應 
JAVA/Hadoop相關的環境變量配置，習慣了PHP的童鞋在這塊可能容易忽略 
啓動Hadoop提示Starting namenodes on []，namenodes爲空，是由於沒有指定ip或端口，修改hadoop/core-site.xml以下

<configuration>
<property>
<name>dfs.namenode.rpc-address</name>
<value>127.0.0.0:9001</value>
</property>
</configuration>

安裝完成後輸入jps能夠查看到NameNode,DataNode等

 

上報和接收

swoole和workerman都有簡單版本實現的數據監控，包括上報，接收，存儲，展現, 主要使用udp上傳(swoole版本已升級爲tcp長鏈接),redis緩存，文件持久化，highcharts展現，能夠做爲思路參考 
swoole-statistics : https://github.com/smalleyes/statistics 
workerman-statistics : https://github.com/walkor/workerman-statistics 
本例使用swoole提供的接口實現UDP傳輸，由於上報數據是必定程度能夠容錯，因此選擇UDP效率優先 
接收數據臨時存儲在Redis中，每隔幾分鐘刷到文件中存儲，文件名按模塊和時間分割存儲，字段|分割(後面與hive對應)

 

數據轉存

建立Hive數據表

根據文件數據格式編寫Hive數據表, TERMINATED BY字段與前面文件字段分隔符想對應 
對錶按日期分區PARTITIONED BY

CREATE TABLE login (
    time int comment '登錄時間', 
    type string comment '類型,email,username,qq等', 
    device string comment '登錄設備，pc,android,ios', 
    ip string comment '登錄ip', 
    uid int comment '用戶id', 
    is_old int comment '是否老用戶'
) 
PARTITIONED BY (
    `date` string COMMENT 'date'
)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '|';

定時(Crontab)建立hadoop分區

hive -e "use web_stat; alter table login add if not exists partition (date='${web_stat_day}')"

轉存

Flume監聽文件目錄，將數據傳輸到能訪問Hdfs集羣的服務器上,這裏傳輸到了224機器的7000端口

#agent3表示代理名稱 login
agent3.sources=source1
agent3.sinks=sink1
agent3.channels=channel1
配置source1

 

配置source1

agent3.sources.source1.type=spooldir
agent3.sources.source1.spoolDir=/data/releases/stat/Data/10001/
agent3.sources.source1.channels=channel1
agent3.sources.source1.fileHeader = false

 

配置sink1

agent3.sinks.sink1.type=avro
agent3.sinks.sink1.hostname=192.168.23.224
agent3.sinks.sink1.port=7000
agent3.sinks.sink1.channel=channel1

 

配置channel1

agent3.channels.channel1.type=file
agent3.channels.channel1.checkpointDir=/data/flume_data/checkpoint_login
agent3.channels.channel1.dataDirs=/data/flume_data/channelData_login

 

啓動flume

加到supervisor守護進程

/home/flume/bin/flume-ng agent -n agent3 -c /home/flume/conf/ -f /home/flume/conf/statistics/login_flume.conf -Dflume.root.logger=info,console

 

224機器監聽7000端口，將數據寫到hdfs集羣

agent1表示代理名稱

agent4.sources=source1
agent4.sinks=sink1
agent4.channels=channel1

 

配置source1

agent4.sources.source1.type=avro
agent4.sources.source1.bind=192.168.23.224
agent4.sources.source1.port=7000
agent4.sources.source1.channels=channel1

 

配置sink1

agent4.sinks.sink1.type=hdfs
agent4.sinks.sink1.hdfs.path=hdfs://hdfs/umr-ubvzlf/uhiveubnhq5/warehouse/web_stat.db/login/date\=%Y-%m-%d
agent4.sinks.sink1.hdfs.fileType=DataStream
agent4.sinks.sink1.hdfs.filePrefix=buffer_census_
agent4.sinks.sink1.hdfs.writeFormat=TEXT
agent4.sinks.sink1.hdfs.rollInterval=30
agent4.sinks.sink1.hdfs.inUsePrefix = .
agent4.sinks.sink1.hdfs.rollSize=536870912
agent4.sinks.sink1.hdfs.useLocalTimeStamp = true
agent4.sinks.sink1.hdfs.rollCount=0
agent4.sinks.sink1.channel=channel1

 

配置channel1

agent4.channels.channel1.type=file
agent4.channels.channel1.checkpointDir=/data/flume_data/login_checkpoint
agent4.channels.channel1.dataDirs=/data/flume_data/login_channelData

 

啓動

加到supervisor守護進程

/usr/local/flume/bin/flume-ng agent -n agent4 -c /usr/local/flume/conf/ -f /usr/local/flume/conf/statistics/login_flume.conf -Dflume.root.logger=info,console

清洗數據

經過Thrift的PHP擴展包調用Hive，編寫類SQL的HQL轉換爲MapReduce任務讀取計算HDFS裏的數據, 將結果存儲在MySQL中 
php-thrift-client下載地址: https://github.com/garamon/php-thrift-hive-client

 

define('THRIFT_HIVE' , ROOT .'/libs/thrift');
$GLOBALS['THRIFT_ROOT'] = THRIFT_HIVE . '/lib';
require_once $GLOBALS['THRIFT_ROOT'] .         '/packages/hive_service/ThriftHive.php';
require_once $GLOBALS['THRIFT_ROOT'] . '/transport/TSocket.php';
require_once $GLOBALS['THRIFT_ROOT'] . '/protocol/TBinaryProtocol.php';
require_once THRIFT_HIVE . '/ThriftHiveClientEx.php';
$transport = new \TSocket('127.0.0.1', 10000);
$transport->setSendTimeout(600 * 1000);
$transport->setRecvTimeout(600 * 1000);
$this->client = new \ThriftHiveClientEx(new \TBinaryProtocol($transport));
$this->client->open();
$this->client->execute("show databases");
$result = $this->client->fetchAll();
var_dump($result);
$this->client->close();

HQL語法說明: https://chu888chu888.gitbooks.io/hadoopstudy/content/Content/8/chapter0803.html

注意的是，儘可能要將HQL語句能轉換爲MapReduce任務，否則沒利用上Hadoop的大數據計算分析，就沒意義 
例以下面的邏輯，取出來在內存裏分析，這樣的邏輯儘可能避免，由於sql在hive裏執行就是普普統統的數據，沒有轉換爲mapreduce

select * from login limit 5;
// php處理
$count = 0;
    foreach ($queryResult as $row) {
      $count ++;
}

一次性轉換爲MapReduce，利用Hadoop的計算能力

select type,count(*) from login group by type;  // 這樣就用到了

建表使用了PARTITIONED BY分區斷言後，查詢就能夠利用分區剪枝（input pruning）的特性，可是斷言字段必須離where關鍵字最近才能被利用上 
// 如前面的login表使用到了date分區斷言，這裏就得把date條件放在第一位

select count(*) from login where date='2016-08-23' and is_old=1;

Hive中不支持等值連表，以下

select * from dual a,dual b where a.key = b.key;

應寫爲：

select * from dual a join dual b on a.key = b.key;

Hive中不支持insert,並且邏輯上也不容許，應爲hadoop是咱們用來作大數據分析，而不該該做爲業務細分數據

 

數據報表展現

這一步就簡單了，讀取MySQL數據，使用highcharts等工具作各類展現，也能夠用crontab定時執行php腳本發送日報，週報等等

 

後續更新

最近看一些資料和別人溝通發現，清洗數據這一步徹底不用php,能夠專一於HQL實現清洗邏輯，將結果保存在hadoop中，再用Sqoop將hadoop數據和MySQL數據同步。即簡化了流程，免去mysql手工插入，又作到了數據更實時，爲二次清洗邏輯的連表HQL作了鋪墊