CUBE SUGAR CONTAINER

技術系のこと書きます。

Apache Hive で圧縮形式のデータを扱う

CentOS7 Apache Hive Python

Apache Hive のテーブルを構成するデータは、デフォルトでは無圧縮になっている。 しかし、設定を変更することで圧縮形式のデータも扱うことができる。 そこで、今回は Apache Hive で圧縮形式のデータを扱ってみることにする。

データを圧縮することには、主に二つのメリットがある。 まず一つ目は HDFS 上のサイズが小さくなるのでディスク容量の節約になること。 そして二つ目こそ本命だけどサイズが小さくなるので読み出しにかかるディスク I/O の負荷も下げることができる。 Hadoop においてディスク I/O は最もボトルネックになりやすいところなので、これは重要となる。

使った環境は次の通り。

$ cat /etc/redhat-release
CentOS Linux release 7.4.1708 (Core) 
$ uname -r
3.10.0-693.11.1.el7.x86_64
$ hive --version
Hive 2.3.2
Git git://stakiar-MBP.local/Users/stakiar/Desktop/scratch-space/apache-hive -r 857a9fd8ad725a53bd95c1b2d6612f9b1155f44d
Compiled by stakiar on Thu Nov 9 09:11:39 PST 2017
From source with checksum dc38920061a4eb32c4d15ebd5429ac8a
$ hadoop version
Hadoop 2.8.3
Subversion https://git-wip-us.apache.org/repos/asf/hadoop.git -r b3fe56402d908019d99af1f1f4fc65cb1d1436a2
Compiled by jdu on 2017-12-05T03:43Z
Compiled with protoc 2.5.0
From source with checksum 9ff4856d824e983fa510d3f843e3f19d
This command was run using /home/vagrant/hadoop-2.8.3/share/hadoop/common/hadoop-common-2.8.3.jar

準備

まずは、ひとまず Hive の CLI を起動しておく。

$ hive

起動したら、次のように set 命令を使って設定を確認しておこう。 hive.exec.compress.output は出力結果を圧縮形式にするかを設定する項目になっている。 そして mapred.output.compression.codec は圧縮に使うコーデックを指定する項目になっている。

hive> set hive.exec.compress.output;
hive.exec.compress.output=false
hive> set mapred.output.compression.codec;
mapred.output.compression.codec=org.apache.hadoop.io.compress.DefaultCodec

上記のように、デフォルトでは出力結果を圧縮しないようになっている。

結果が圧縮されないことを確認したところで、まずは無圧縮のテーブルを作ってみよう。 以下の設定ではデータが CSV で保存されることになる。

hive> CREATE TABLE users (
    >   name STRING,
    >   age INT
    > )
    > ROW FORMAT DELIMITED
    > FIELDS TERMINATED BY ','
    > STORED AS TEXTFILE;
OK
Time taken: 12.171 seconds

レコードを追加する。

hive> INSERT INTO TABLE users VALUES
    >   ("Alice", 20),
    >   ("Bob", 25),
    >   ("Carol", 30);
...
OK
Time taken: 40.852 seconds

この状態で、テーブルを構成するデータがどのように保存されているかをまずは確認しておこう。 テーブルの保存先は SHOW CREATE TABLE で確認できる。

hive> SHOW CREATE TABLE users;
OK
CREATE TABLE `users`(
  `name` string, 
  `age` int)
ROW FORMAT SERDE 
  'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe' 
WITH SERDEPROPERTIES ( 
  'field.delim'=',', 
  'serialization.format'=',') 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.mapred.TextInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION
  'hdfs://master:9000/user/hive/warehouse/users'
TBLPROPERTIES (
  'transient_lastDdlTime'='1518700567')
Time taken: 0.203 seconds, Fetched: 16 row(s)

上記の LOCATION が保存先を表している。

別のターミナルで hdfs dfs -ls コマンドを使って上記のパスを確認してみよう。

$ hdfs dfs -ls /user/hive/warehouse/users
Found 1 items
-rwxrwxr-x   2 vagrant supergroup         25 2018-02-15 13:16 /user/hive/warehouse/users/000000_0

ディレクトリの中には 000000_0 というファイルだけがあることが分かる。

hdfs dfs -cat コマンドを使って内容を確認してみよう。

$ hdfs dfs -cat /user/hive/warehouse/users/000000_0
Alice,20
Bob,25
Carol,30

すると、これが CSV ファイルで先ほど投入したレコードが入っていることが分かる。 デフォルトでは、このようにファイルが無圧縮で HDFS 上にそのまま保存されることになる。

テーブルを構成するデータを圧縮形式にする

続いてはテーブルを構成するデータを圧縮したものにしてみよう。 それには、前述した設定項目を編集する。

hive> set hive.exec.compress.output=true;
hive> set mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec;

ここではデータを GZIP で圧縮するように設定した。

先ほど作った無圧縮のテーブルから、新しいテーブルを作ってみる。 テーブル自体の設定自体は先ほどと変えておらず中身は CSV になるよう指定している。

hive> CREATE TABLE users_gzip
    >   ROW FORMAT DELIMITED
    >   FIELDS TERMINATED BY ','
    >   STORED AS TEXTFILE
    > AS SELECT * FROM users;
...
OK
Time taken: 28.895 seconds

先ほどと同じように保存先の HDFS を確認してみよう。

$ hdfs dfs -ls /user/hive/warehouse/users_gzip
Found 1 items
-rwxrwxr-x   2 vagrant supergroup         45 2018-02-15 13:22 /user/hive/warehouse/users_gzip/000000_0.gz

すると、今度はディレクトリの中にあるファイルに .gz という名前がついていることが分かる。

ファイルをローカルにダウンロードしてこよう。

$ hdfs dfs -get /user/hive/warehouse/users_gzip/000000_0.gz

file コマンドを使って形式を確認すると、ちゃんと GZIP ファイルとなっている。

$ file 000000_0.gz
000000_0.gz: gzip compressed data, from Unix

gunzip コマンドを使って圧縮ファイルを解凍してみよう。

$ gunzip 000000_0.gz

解凍したファイルを確認すると CSV ファイルになっていることが分かる。

$ cat 000000_0 
Alice,20
Bob,25
Carol,30

このように設定を切り替えることでテーブルを構成するデータを圧縮形式にできる。

圧縮形式によるサイズの違いを比べてみる

続いては圧縮形式を変えることでテーブルのサイズがどのように変化するか見てみる。 違いを確かめるには、ある程度のサイズが必要なのでダミーデータを作ることにした。

ダミーデータの生成には Python の faker を使うことにする。 pip コマンドを使ってインストールしよう。

$ sudo yum -y install epel-release
$ sudo yum -y install python-pip
$ sudo pip install faker

インストールができたら Python の REPL を起動する。

$ python

次のようにして 10 万件のダミーデータが入った CSV ファイルを作る。

>>> from faker import Faker
>>> fake = Faker()
>>> import random
>>> N = 100000
>>> with open('users.csv', 'w') as f:
...     for _ in range(N):
...         age = random.randint(0, 100)
...         name = fake.last_name()
...         f.write(name + ',' + str(age) + '\n')
...

こんな感じでダミーデータの入った CSV ファイルができる。

$ wc -l users.csv
100000 users.csv
$ head users.csv 
Moore,82
Jensen,40
Robinson,42
White,11
Atkinson,56
Small,17
Wilson,76
Johnson,64
Moody,85
Barnes,61

それを Hive のテーブルに取り込む。

hive> LOAD DATA LOCAL INPATH 'users.csv' OVERWRITE INTO TABLE users;
Loading data to table default.users
OK
Time taken: 1.026 seconds

上手くいけば次のように 10 万件のデータが見えるようになる。

hive> SELECT COUNT(1) FROM users;
...
OK
100000
Time taken: 39.95 seconds, Fetched: 1 row(s)

GZIP

まずは GZIP 形式から。 先ほど作ったテーブルは一旦消しておこう。

DROP TABLE users_gzip;

圧縮に使うコーデックとして GZIP を指定する。

hive> set hive.exec.compress.output=true;
hive> set mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec;

先ほど作った 10 万件のテーブルをコピーして新しいテーブルを作る。

hive> CREATE TABLE users_gzip
    >   ROW FORMAT DELIMITED
    >   FIELDS TERMINATED BY ','
    >   STORED AS TEXTFILE
    > AS SELECT * FROM users;
...
OK
Time taken: 27.286 seconds

この通りデータがコピーされた。

hive> SELECT COUNT(1) FROM users_gzip;
OK
100000
Time taken: 0.207 seconds, Fetched: 1 row(s)

テーブルのサイズは hdfs dfs -du コマンドを使って確認できる。

$ hdfs dfs -du -h /user/hive/warehouse | grep users
974.5 K  /user/hive/warehouse/users
335.9 K  /user/hive/warehouse/users_gzip

GZIP で圧縮するとテーブルのサイズが約 34% まで小さくなった。

BZIP2

同じことを BZIP2 でもやる。

hive> set hive.exec.compress.output=true;
hive> set mapred.output.compression.codec=org.apache.hadoop.io.compress.BZip2Codec;
hive> CREATE TABLE users_bzip2
    >   ROW FORMAT DELIMITED
    >   FIELDS TERMINATED BY ','
    >   STORED AS TEXTFILE
    > AS SELECT * FROM users;
...
OK
Time taken: 27.634 seconds

Snappy

続いては Snappy で。

hive> set hive.exec.compress.output=true;
hive> set mapred.output.compression.codec=org.apache.hadoop.io.compress.SnappyCodec;
hive> CREATE TABLE users_snappy
    >   ROW FORMAT DELIMITED
    >   FIELDS TERMINATED BY ','
    >   STORED AS TEXTFILE
    > AS SELECT * FROM users;
...
OK
Time taken: 26.979 seconds

LZ4

最後に LZ4 を。

hive> set hive.exec.compress.output=true;
hive> set mapred.output.compression.codec=org.apache.hadoop.io.compress.Lz4Codec;
hive> CREATE TABLE users_lz4
    >   ROW FORMAT DELIMITED
    >   FIELDS TERMINATED BY ','
    >   STORED AS TEXTFILE
    > AS SELECT * FROM users;
...
OK
Time taken: 26.433 seconds

圧縮率を比較する

これで一通り出揃ったのでサイズを確認してみよう。 ちなみに圧縮に使えるコーデックは Hadoop のバージョンによって異なる。

$ hdfs dfs -du -h /user/hive/warehouse | grep users
974.5 K  /user/hive/warehouse/users
228.6 K  /user/hive/warehouse/users_bzip2
335.9 K  /user/hive/warehouse/users_gzip
583.6 K  /user/hive/warehouse/users_lz4
603.8 K  /user/hive/warehouse/users_snappy

上記から圧縮したときのサイズは BZIP2 < GZIP < LZ4 < Snappy ということが分かった。 ただし、圧縮率の高いコーデックは解凍に時間がかかる。 つまり、ディスク I/O の負担は減るが、代わりに CPU の負担が増えることになる。

また、圧縮形式を決める上では、その形式が「スプリット可能」かについても注意する必要があるらしい。 スプリット可能というのは、大きなファイルを分割して複数のマッパー、リデューサーで処理できることを表している。 前述した圧縮形式の中では BZIP2 についてはスプリット可能、LZ4 と GZIP と Snappy はスプリット不能となっている。 スプリット不能な圧縮形式では、大きなファイルも一つのマッパー、リデューサーで処理しなければならない。 そのため、分散処理のメリットが薄れてしまう。

ただし、上記はあくまで「大きなファイルがある」ことが前提になっている。 つまり、テーブルを構成するファイルが全て適度な大きさになっていればスプリット不能でも問題にはならない。 また GZIP や Snappy といった圧縮形式がスプリット不能なのは、あくまでテキストファイルで保存する場合に限られるようだ。 後述する Sequence ファイルや ORC ファイルといったフォーマットで保存するなら、スプリット可能になるらしい。

一つのテーブルをまぜこぜの圧縮形式で構成してみる

ここまでやってきて、ちょっとした疑問が浮かぶ。 圧縮形式を Hive の設定項目で切り替えるということは、テーブル自体にはそのメタ情報が保存されていないことを意味する。 だとすれば、一つのテーブルを複数の圧縮形式が入り乱れた状態で作ることもできるのだろうか。 先に結論から書くと、これはできるようになっている。

検証用のテーブルを用意する。

hive> CREATE TABLE users_mixed (
    >   name STRING,
    >   age INT
    > )
    > ROW FORMAT DELIMITED
    > FIELDS TERMINATED BY ','
    > STORED AS TEXTFILE;
OK
Time taken: 0.091 seconds

圧縮に使うコーデックの設定を切り替えながらレコードを追加する。

hive> set hive.exec.compress.intermediate=true;
hive> set mapred.output.compression.codec=org.apache.hadoop.io.compress.SnappyCodec;
hive> INSERT INTO TABLE users_mixed VALUES ("Alice", 20);
...
hive> set mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec;
hive> INSERT INTO TABLE users_mixed VALUES ("Bob", 25);
...
hive> set mapred.output.compression.codec=org.apache.hadoop.io.compress.BZip2Codec;
hive> INSERT INTO TABLE users_mixed VALUES ("Carol", 30);

レコードを追加した後、テーブルを構成するファイルを確認すると、次のような結果が得られる。 それぞれのファイルの末尾には異なる拡張子がついていて、複数の圧縮形式が混在している状況だ。

$ hdfs dfs -ls /user/hive/warehouse/users_mixed
Found 3 items
-rwxrwxr-x   2 vagrant supergroup         54 2018-02-15 13:49 /user/hive/warehouse/users_mixed/000000_0.bz2
-rwxrwxr-x   2 vagrant supergroup         27 2018-02-15 13:48 /user/hive/warehouse/users_mixed/000000_0.gz
-rwxrwxr-x   2 vagrant supergroup         19 2018-02-15 13:48 /user/hive/warehouse/users_mixed/000000_0.snappy

この状態であっても、テーブルからはちゃんとレコードを読み出せる。

hive> SELECT * FROM users_mixed;
OK
Carol   30
Bob 25
Alice   20
Time taken: 0.271 seconds, Fetched: 3 row(s)

どうしてこんなことができるかというと、拡張子からファイルの種類を自動的に推定して処理してくれるようになっているため。 ただし、この処理方法はデータを保存するフォーマットがテキストファイルの場合だから、という点に注意が必要となる。 後述する別のフォーマットでは、ファイルの種類に関する情報がファイル自体やテーブルのメタデータに書き込まれることになる。 そのため、まぜこぜにできない場合もある。

Sequence ファイルを圧縮してみる

ここまでの例では、全てファイルのフォーマットとしてテキストファイルを使っていた。 つまり、テーブル定義で STORED AS TEXTFILE としていた。 ここからは別のフォーマットを使った場合にどうなるか確認してみよう。 テキストファイルを使ったときとは、少し勝手が異なる場合があるようだ。

最初は Sequence ファイルというフォーマットを試してみよう。 このフォーマットはバイナリ形式になっている。 Sequence ファイルのフォーマットであれば、テキストファイルでスプリット不能だった圧縮形式でもスプリット可能になる。

まずは設定で GZIP を使った圧縮を有効にしておく。

hive> set hive.exec.compress.output=true;
hive> set mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec;

続いて Sequence ファイルを使ってデータを保存するようにしたテーブルを定義する。 具体的にはテーブルを定義する時点で STORED AS SEQUENCEFILE とする。

hive> CREATE TABLE users_sequence (
    >   name STRING,
    >   age INT
    > )
    > ROW FORMAT DELIMITED
    > FIELDS TERMINATED BY ','
    > STORED AS SEQUENCEFILE;
OK
Time taken: 0.07 seconds

作成したテーブルにレコードを追加してみよう。

hive> INSERT INTO TABLE users_sequence VALUES ("Alice", 20);
...
OK
Time taken: 27.658 seconds

レコードが追加できたら HDFS 上でどのようになっているか確認する。

$ hdfs dfs -ls /user/hive/warehouse/users_sequence
Found 1 items
-rwxrwxr-x   2 vagrant supergroup        168 2018-02-15 14:07 /user/hive/warehouse/users_sequence/000000_0

すると、先ほどのテキストファイルの場合とは異なりファイルに .gz の拡張子がついていない。

バイナリフォーマットのファイルなので、ちょっと乱暴だけど内容をテキストとして表示してみよう。

$ hdfs dfs -cat /user/hive/warehouse/users_sequence/000000_0
SEQ"org.apache.hadoop.io.BytesWritableorg.apache.hadoop.io.Text'org.apache.hadoop.io.compress.GzipCodec41e?:???Qt?j!!?p??LN?123ݟ?

すると、ファイルの中に圧縮に使ったコーデックの情報が含まれていることが分かる。 このように Sequence ファイルでは、ファイル自体に圧縮形式の情報が含まれることになる。 また、このやり方ならファイルを丸ごと圧縮する方法ではないためスプリット可能になる理由もうなずける。 ちゃんとレコードの区切りを考えて各パートごとに圧縮した情報をメタデータとしてファイルに残しておけるからだ。

ちなみに Sequence ファイルについては hdfs dfs -text コマンドで中身が確認できるようになっている。

$ hdfs dfs -text /user/hive/warehouse/users_sequence/000000_0
18/02/15 14:16:13 INFO zlib.ZlibFactory: Successfully loaded & initialized native-zlib library
18/02/15 14:16:13 INFO compress.CodecPool: Got brand-new decompressor [.gz]
    Alice,20

一応 Snappy で圧縮したときの結果も確認しておこう。

hive> set mapred.output.compression.codec=org.apache.hadoop.io.compress.SnappyCodec;
hive> INSERT INTO TABLE users_sequence VALUES ("Bob", 25);
...
OK
Time taken: 29.17 seconds

さっきと同じようにファイルの中に圧縮に使ったコーデックの情報が記録されていることが分かる。

$ hdfs dfs -cat /user/hive/warehouse/users_sequence/000000_0_copy_1
SEQ"org.apache.hadoop.io.BytesWritableorg.apache.hadoop.io.Text)org.apache.hadoop.io.compress.SnappyCodech???7?
                                                                                                               ??h?   Bob,25

ORC

次は ORC (The Optimized Row Columnar) ファイルについて。 ORC ファイルは Sequence ファイルと同じようにバイナリ形式のフォーマットになっている。 Hive に最適化されたカラム志向型のデータフォーマットになっており高速に動作する。 いくつかの資料を見ると、この ORC ファイルを使うのが Hive のベストプラクティスのようだ。

jp.hortonworks.com

LanguageManual ORC - Apache Hive - Apache Software Foundation

ORC ファイルを圧縮するときは、これまでとはまたちょっと勝手が違っている。 何かというと、テーブル自体のメタデータに圧縮形式が記録されるため。 つまり、これまで使ってきた設定項目の hive.exec.compress.output などの内容は無視されるので注意が必要となる。

それでは ORC ファイルでデータを保存するテーブルを定義しよう。 実は、特に指定はないけどこれだけでファイルが GZIP で圧縮されるようにデフォルト値がなっている。

hive> CREATE TABLE users_orc (
    >   name STRING,
    >   age INT
    > )
    > ROW FORMAT DELIMITED
    > FIELDS TERMINATED BY ','
    > STORED AS ORC;
OK
Time taken: 0.143 seconds

レコードを追加してみよう。

hive> INSERT INTO TABLE users_orc VALUES ("Alice", 20);
...
OK
Time taken: 24.535 seconds

今回も、ファイルを確認すると拡張子には何もついていない。

$ hdfs dfs -ls /user/hive/warehouse/users_orc/
Found 1 items
-rwxrwxr-x   2 vagrant supergroup        295 2018-02-15 14:24 /user/hive/warehouse/users_orc/000000_0

中身を確認すると ORC という文字列が見えるため、どうやら確かに ORC ファイルのようだ。

$ hdfs dfs -cat /user/hive/warehouse/users_orc/000000_0
ORC
P8??be!1F%.Vǜ??T%??+


(((P
    AliceFPN(V?b?``???ь?`?IBH3?@? H???LlBL
                                                @??T???b?`
        ???`hg???`p?Q??`T??bbd?b?K?M?bNLOUb?`bfF+?+1F%.Vǜ??T%???`b??А`p?[??"!6
                                                                            (-0??ORC

続いては、どのように ORC ファイルのテーブルで圧縮形式を指定するか見ていこう。 その前に、一旦先ほど作ったテーブルは削除しておく。

hive> DROP TABLE users_orc;
OK
Time taken: 0.195 seconds

結論から先に書くと ORC ファイルでは TBLPROPERTIES という書式を用いて圧縮形式を指定する。 この中にキーバリュー形式で orc.compress に圧縮形式を指定する。 以下では SNAPPY を指定している。 次のクエリではダミーデータの入ったテーブルからレコードをコピーしている。

hive> CREATE TABLE users_orc_snappy
    >   ROW FORMAT DELIMITED
    >   FIELDS TERMINATED BY ','
    >   STORED AS ORC TBLPROPERTIES ("orc.compress" = "SNAPPY")
    > AS SELECT * FROM users;

同じように、指定なし、ZLIB、NONE を指定したものを作っていこう。

hive> CREATE TABLE users_orc
    >   ROW FORMAT DELIMITED
    >   FIELDS TERMINATED BY ','
    >   STORED AS ORC
    > AS SELECT * FROM users;
...
hive> CREATE TABLE users_orc_gzip
    >   ROW FORMAT DELIMITED
    >   FIELDS TERMINATED BY ','
    >   STORED AS ORC TBLPROPERTIES ("orc.compress" = "ZLIB")
    > AS SELECT * FROM users;
...
hive> CREATE TABLE users_orc_none
    >   ROW FORMAT DELIMITED
    >   FIELDS TERMINATED BY ','
    >   STORED AS ORC TBLPROPERTIES ("orc.compress" = "NONE")
    > AS SELECT * FROM users;

結果は次の通り。 指定しなかったものと GZIP が同じサイズになっており、デフォルトで圧縮が効いていることが確認できる。 今回のデータでは、圧縮をかけない NONE と Snappy のサイズの違いがほとんど出なかった。

$ hdfs dfs -du -h /user/hive/warehouse | grep users_orc
231.6 K  /user/hive/warehouse/users_orc
231.6 K  /user/hive/warehouse/users_orc_gzip
301.2 K  /user/hive/warehouse/users_orc_none
300.1 K  /user/hive/warehouse/users_orc_snappy

まとめ

今回は Apache Hive のテーブルを構成するファイルを圧縮する方法について扱った。 ファイルを圧縮すると、ディスク容量の節約やパフォーマンスの向上が見込める。 ただし、ファイルを圧縮するときは、その特性やファイルフォーマットとの相性について理解を深める必要もある。 例えば、テキストファイルでは圧縮形式によってスプリット可能・不能といった問題が出てくる。 また、フォーマットごとに圧縮形式の情報を何処に残すかが異なっていたり、使える形式についても違ったりすることも分かった。

プログラミング Hive

プログラミング Hive

  • 作者: Edward Capriolo,Dean Wampler,Jason Rutherglen,佐藤直生,嶋内翔,Sky株式会社玉川竜司
  • 出版社/メーカー: オライリージャパン
  • 発売日: 2013/06/15
  • メディア: 大型本
  • この商品を含むブログ (3件) を見る

スマートPythonプログラミング: Pythonのより良い書き方を学ぶ

スマートPythonプログラミング: Pythonのより良い書き方を学ぶ