自從Hadoop集群搭建以來，我們一直使用的是Gzip進行壓縮

當時，我對gzip壓縮過的文件和原始的log文件分別跑MapReduce測試，最終執(zhí)行速度基本差不多

而且Hadoop原生支持Gzip解壓，所以，當時就直接采用了Gzip壓縮的方式

關于Lzo壓縮，twitter有一篇文章，介紹的比較詳細，見這里：

Lzo壓縮相比Gzip壓縮，有如下特點：

1. 壓縮解壓的速度很快
2. Lzo壓縮是基于Block分塊的，這樣，一個大的文件（在Hadoop上可能會占用多個Block塊），就可以由多個MapReduce并行來進行處理

雖然Lzo的壓縮比沒有Gzip高，不過由于其前2個特性，在Hadoop上使用Lzo還是能整體提升集群的性能的

我測試了12個log文件，總大小為8.4G，以下是Gzip和Lzo壓縮的結果：

1. Gzip壓縮，耗時480s，Gunzip解壓，耗時180s，壓縮后大小為2.5G
2. Lzo壓縮，耗時160s，Lzop解壓，耗時110s，壓縮后大小為4G

以下為在Hadoop集群上使用Lzo的步驟：

1. 在集群的所有節(jié)點上安裝Lzo庫，可從這里下載

cd /opt/ysz/src/lzo-2.04

./configure –enable-shared

make

make install

#編輯/etc/ld.so.conf，加入/usr/local/lib/后，執(zhí)行/sbin/ldconfig

或者cp /usr/local/lib/liblzo2.* /usr/lib64/

#如果沒有這一步，最終會導致以下錯誤：

lzo.LzoCompressor: java.lang.UnsatisfiedLinkError: Cannot load liblzo2.so.2 (liblzo2.so.2: cannot open shared object file: No such file or directory)!

2. 編譯安裝Hadoop Lzo本地庫以及Jar包，從這里下載

    export CFLAGS=-m64

    export CXXFLAGS=-m64

    ant compile-native tar

    #將本地庫以及Jar包拷貝到hadoop對應的目錄下，并分發(fā)到各節(jié)點上

    cp lib/native/Linux-amd64-64/* /opt/sohuhadoop/hadoop/lib/native/Linux-amd64-64/

    cp hadoop-lzo-0.4.10.jar /opt/sohuhadoop/hadoop/lib/

3. 設置Hadoop，啟用Lzo壓縮

vi core-site.xml

<property>     

<name>io.compression.codecs</name>     

<value>org.apache.hadoop.io.compress.GzipCodec,org.apache.hadoop.io.compress.DefaultCodec,com.hadoop.compression.lzo.LzoCodec,com.hadoop.compression.lzo.LzopCodec,org.apache.hadoop.io.compress.BZip2Codec</value>

</property>

<property>     

<name>io.compression.codec.lzo.class</name>     

<value>com.hadoop.compression.lzo.LzoCodec</value>

</property>

 

vi mapred-site.xml

<property>

<name>mapred.compress.map.output</name>     

<value>true</value>   

</property>   

<property>     

<name>mapred.map.output.compression.codec</name>      

<value>com.hadoop.compression.lzo.LzoCodec</value>   

</property>

4. 安裝lzop，從這里下載

下面就是使用lzop壓縮log文件，并上傳到Hadoop上，執(zhí)行MapReduce操作，測試的Hadoop是由3個節(jié)點組成集群

lzop -v 2011041309.log

hadoop fs -put *.lzo /user/pvlog

#給Lzo文件建立Index

hadoop jar /opt/sohuhadoop/hadoop/lib/hadoop-lzo-0.4.10.jar com.hadoop.compression.lzo.LzoIndexer /user/pvlog/

寫一個簡單的MapReduce來測試，需要指定InputForamt為Lzo格式，否則對單個Lzo文件仍不能進行Map的并行處理

job.setInputFormatClass(com.hadoop.mapreduce.LzoTextInputFormat.class);

可以通過下面的代碼來設置Reduce的數(shù)目：

job.setNumReduceTasks(8);

最終，12個文件被切分成了36個Map任務來并行處理，執(zhí)行時間為52s，如下圖：

我們配置Hadoop默認的Block大小是128M，如果我們想切分成更多的Map任務，可以通過設置其最大的SplitSize來完成：

FileInputFormat.setMaxInputSplitSize(job, 64 *1024 * 1024);

最終，12個文件被切分成了72個Map來處理，但處理時間反而長了，為59s，如下圖：

而對于Gzip壓縮的文件，即使我們設置了setMaxInputSplitSize，最終的Map數(shù)仍然是輸入文件的數(shù)目12，執(zhí)行時間為78s，如下圖：

從以上的簡單測試可以看出，使用Lzo壓縮，性能確實比Gzip壓縮要好不少

最近在看《Hadoop:The Definitive Guide》，對其分布式文件系統(tǒng)HDFS的Streaming data access不能理解。基于流的數(shù)據(jù)讀寫，太抽象了，什么叫基于流，什么是流？Hadoop是Java語言寫的，所以想理解好Hadoop的Streaming Data Access，還得從Java流機制入手。流機制也是JAVA及C++中的一個重要的機制，通過流使我們能夠自由地操作包括文件，內存，IO設備等等中的數(shù)據(jù)。

首先，流是什么？

流是個抽象的概念，是對輸入輸出設備的抽象，Java程序中，對于數(shù)據(jù)的輸入/輸出操作都是以“流”的方式進行。設備可以是文件，網(wǎng)絡，內存等。

流具有方向性，至于是輸入流還是輸出流則是一個相對的概念，一般以程序為參考，如果數(shù)據(jù)的流向是程序至設備，我們成為輸出流，反之我們稱為輸入流。

可以將流想象成一個“水流管道”，水流就在這管道中形成了，自然就出現(xiàn)了方向的概念。

當程序需要從某個數(shù)據(jù)源讀入數(shù)據(jù)的時候，就會開啟一個輸入流，數(shù)據(jù)源可以是文件、內存或網(wǎng)絡等等。相反地，需要寫出數(shù)據(jù)到某個數(shù)據(jù)源目的地的時候，也會開啟一個輸出流，這個數(shù)據(jù)源目的地也可以是文件、內存或網(wǎng)絡等等。

流有哪些分類？

可以從不同的角度對流進行分類：

1. 處理的數(shù)據(jù)單位不同，可分為：字符流，字節(jié)流

2.數(shù)據(jù)流方向不同，可分為：輸入流，輸出流

3.功能不同，可分為：節(jié)點流，處理流

1. 和 2. 都比較好理解，對于根據(jù)功能分類的，可以這么理解：

節(jié)點流：節(jié)點流從一個特定的數(shù)據(jù)源讀寫數(shù)據(jù)。即節(jié)點流是直接操作文件，網(wǎng)絡等的流，例如FileInputStream和FileOutputStream，他們直接從文件中讀取或往文件中寫入字節(jié)流。

處理流：“連接”在已存在的流（節(jié)點流或處理流）之上通過對數(shù)據(jù)的處理為程序提供更為強大的讀寫功能。過濾流是使用一個已經(jīng)存在的輸入流或輸出流連接創(chuàng)建的，過濾流就是對節(jié)點流進行一系列的包裝。例如BufferedInputStream和BufferedOutputStream，使用已經(jīng)存在的節(jié)點流來構造，提供帶緩沖的讀寫，提高了讀寫的效率，以及DataInputStream和DataOutputStream，使用已經(jīng)存在的節(jié)點流來構造，提供了讀寫Java中的基本數(shù)據(jù)類型的功能。他們都屬于過濾流。

舉個簡單的例子：

public static void main(String[] args) throws IOException {
        // 節(jié)點流FileOutputStream直接以A.txt作為數(shù)據(jù)源操作
        FileOutputStream fileOutputStream = new FileOutputStream("A.txt");
        // 過濾流BufferedOutputStream進一步裝飾節(jié)點流，提供緩沖寫
        BufferedOutputStream bufferedOutputStream = new BufferedOutputStream(
                fileOutputStream);
        // 過濾流DataOutputStream進一步裝飾過濾流，使其提供基本數(shù)據(jù)類型的寫
        DataOutputStream out = new DataOutputStream(bufferedOutputStream);
        out.writeInt(3);
        out.writeBoolean(true);
        out.flush();
        out.close();
        // 此處輸入節(jié)點流，過濾流正好跟上邊輸出對應，讀者可舉一反三
        DataInputStream in = new DataInputStream(new BufferedInputStream(
                new FileInputStream("A.txt")));
        System.out.println(in.readInt());
        System.out.println(in.readBoolean());
        in.close();
}

流結構介紹：

Java所有的流類位于java.io包中，都分別繼承字以下四種抽象流類型。

1.繼承自InputStream/OutputStream的流都是用于向程序中輸入/輸出數(shù)據(jù)，且數(shù)據(jù)的單位都是字節(jié)(byte=8bit)，如圖，深色的為節(jié)點流，淺色的為處理流。

2.繼承自Reader/Writer的流都是用于向程序中輸入/輸出數(shù)據(jù)，且數(shù)據(jù)的單位都是字符(2byte=16bit)，如圖，深色的為節(jié)點流，淺色的為處理流。

常見流類介紹：

節(jié)點流類型常見的有：

對文件操作的字符流有FileReader/FileWriter，字節(jié)流有FileInputStream/FileOutputStream。

處理流類型常見的有：

緩沖流：緩沖流要“套接”在相應的節(jié)點流之上，對讀寫的數(shù)據(jù)提供了緩沖的功能，提高了讀寫效率，同事增加了一些新的方法。

　　字節(jié)緩沖流有BufferedInputStream/BufferedOutputStream，字符緩沖流有BufferedReader/BufferedWriter，字符緩沖流分別提供了讀取和寫入一行的方法ReadLine和NewLine方法。

　　對于輸出地緩沖流，寫出的數(shù)據(jù)，會先寫入到內存中，再使用flush方法將內存中的數(shù)據(jù)刷到硬盤。所以，在使用字符緩沖流的時候，一定要先flush，然后再close，避免數(shù)據(jù)丟失。

轉換流：用于字節(jié)數(shù)據(jù)到字符數(shù)據(jù)之間的轉換。

　　僅有字符流InputStreamReader/OutputStreamWriter。其中，InputStreamReader需要與InputStream“套接”，OutputStreamWriter需要與OutputStream“套接”。

數(shù)據(jù)流：提供了讀寫Java中的基本數(shù)據(jù)類型的功能。

　　DataInputStream和DataOutputStream分別繼承自InputStream和OutputStream，需要“套接”在InputStream和OutputStream類型的節(jié)點流之上。

對象流：用于直接將對象寫入寫出。

　　流類有ObjectInputStream和ObjectOutputStream，本身這兩個方法沒什么，但是其要寫出的對象有要求，該對象必須實現(xiàn)Serializable接口，來聲明其是可以序列化的。否則，不能用對象流讀寫。

　　還有一個關鍵字比較重要，transient，由于修飾實現(xiàn)了Serializable接口的類內的屬性，被該修飾符修飾的屬性，在以對象流的方式輸出的時候，該字段會被忽略。

Hive:

利用squirrel-sql 連接hive

add driver -> name&example url(jdbc:hive2://xxx:10000)->extra class path ->Add

{hive/lib/hive-common-*.jar

hive/lib/hive-contrib-*.jar

hive/lib/hive-jdbc-*.jar

hive/lib/libthrift-*.jar

hive/lib/hive-service-*.jar

hive/lib/httpclient-*.jar

hive/lib/httpcore-*.jar

hadoop/share/hadoop/common/hadoop-common--*.jar

hadoop/share/hadoop/common/lib/common-configuration-*.jar

hadoop/share/hadoop/common/lib/log4j-*.jar

hadoop/share/hadoop/common/lib/slf4j-api-*.jar

hadoop/share/hadoop/common/lib/slf4j-log4j-*.jar}

->List Drivers(wait ..then class name will auto set org.apache.hive.jdbc/HiveDriver)->OK->Add aliases ->chose the hive driver->done

Hive數(shù)據(jù)遷移

1.導出表

EXPORT TABLE <table_name> TO 'path/to/hdfs';

2.復制數(shù)據(jù)到另一個hdfs

hadoop distcp hdfs://:8020/path/to/hdfs hdfs:///path/to/hdfs

3.導入表

IMPORT TABLE <table_name> FROM 'path/to/another/hdfs';

Hive 輸出查詢結果到文件

輸出到本地文件：

insert overwrite local directory './test-04'
row format delimited
FIELDS TERMINATED BY '\t'
COLLECTION ITEMS TERMINATED BY ','
MAP KEYS TERMINATED BY ':'
select * from src;

輸出到hdfs：

輸出到hdfs好像不支持 row format，只能另辟蹊徑了

INSERT OVERWRITE DIRECTORY '/outputable.txt'
select concat(col1, ',', col2, ',', col3) from myoutputtable;

當然默認的分隔符是\001

若要直接對文件進行操作課直接用stdin的形式

eg. hadoop fs -cat ../000000_0 |python doSomeThing.py

#!/usr/bin/env python

import sys

for line in sys.stdin:

    (a,b,c)=line.strip().split('\001')

Hive 語法：

hive好像不支持select dicstinct col1 as col1 from table group by col1

需要用grouping sets

select col1 as col1 from table group by col1 grouping sets((col1))

Beeline:

文檔：https://cwiki.apache.org/confluence/display/Hive/HiveServer2+Clients

利用jdbc連接hive：

hive2='JAVA_HOME=/opt/java7 HADOOP_HOME=/opt/hadoop /opt/hive/bin/beeline -u jdbc:hive2://n1.hd2.host.dxy:10000 -n hadoop -p fake -d org.apache.hive.jdbc.HiveDriver --color=true --silent=false --fastConnect=false --verbose=true'

beeline利用jdbc連接hive若需要執(zhí)行多條命令使用

hive2 -e "xxx" -e "yyy"  -e...