PySpark: Jupyter Notebook からローカルの PySpark ドライバを操作する

今回はローカルで動作している PySpark のドライバプログラムを Jupyter Notebook から操作する方法について。 ようするに Jupyter Notebook と PySpark (のドライバ) が同じマシン上で動く場合ということ。 この場合の設定内容はとてもシンプルで、環境変数を書き換えるだけで使えるようになる。

これがもし Jupyter Notebook と PySpark のホストが分かれていると、もうちょっと複雑になる。 具体的には Apache Ivy と SparkMagic というソフトウェアを連携させないといけない。 それについては今後改めて書くつもり。

使った環境は次の通り。

$ cat /etc/redhat-release
CentOS Linux release 7.4.1708 (Core)
$ uname -r
3.10.0-693.11.1.el7.x86_64
$ pyspark --version
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.2.1
      /_/
                        
Using Scala version 2.11.8, OpenJDK 64-Bit Server VM, 1.8.0_161
Branch 
Compiled by user felixcheung on 2017-11-24T23:19:45Z
Revision 
Url 
Type --help for more information.
$ python3 -V
Python 3.6.4

ドライバのホストに Jupyter Notebook をインストールする

まずは PySpark のドライバを動かしているホストに Jupyter Notebook をインストールする。 どうせ後から必要になるので pandas とかも入れておくと良いかも。

$ sudo pip3 install jupyter pandas

PySpark が Jupyter Notebook を起動するように環境変数を設定する

続いては PySpark を使うときインタプリタとして Jupyter Notebook を使うように環境変数を設定する。

$ cat << 'EOF' >> ~/.bashrc
export PYSPARK_PYTHON=/usr/bin/python3
export PYSPARK_DRIVER_PYTHON=jupyter
export PYSPARK_DRIVER_PYTHON_OPTS='notebook' pyspark
EOF
$ source ~/.bashrc

後は通常通り PySpark を起動すれば、同時に Jupyter Notebook が使えるようになる。 この環境では YARN を使って Spark クラスタを構築した。 そのため --master に yarn を指定している。

$ pyspark --master yarn

あとはブラウザで Jupyter Notebook が動作している場所を開くだけ。

$ open http://localhost:8888

PySpark のドライバがリモートで動作している場合

作業しているホストがローカルホストでないときは PYSPARK_DRIVER_PYTHON_OPTS で Listen するアドレスを指定する。 任意のアドレスで Listen するときは、次のように --ip に 0.0.0.0 を指定する。

$ cat << 'EOF' >> ~/.bashrc
export PYSPARK_PYTHON=/usr/bin/python3.6
export PYSPARK_DRIVER_PYTHON=jupyter
export PYSPARK_DRIVER_PYTHON_OPTS='notebook --ip=0.0.0.0' pyspark
EOF

先ほど紹介したのと同じように起動する。

$ pyspark --master yarn

起動するとコンソールにトークンを含む URL が表示されるはず。

表示された URL を、必要に応じて IP アドレス部分を書き換えたらブラウザで開くだけ。

$ open http://192.168.33.10:8888/?token=XXXXX...

動作確認

Jupyter Notebook の画面が開いたら新しいノートブックを作成して、ちゃんと PySpark が動くか確認しよう。

例えば SparkContext がちゃんと使えるか、とか。

In:
sc

Out:
SparkContext

Spark UI

Version
v2.2.1
Master
yarn
AppName
PySparkShell

分散処理の部分がちゃんと動くかを確かめるためにワードカウントしてみたりね。

In:
rdd = sc.parallelize(['A', 'B', 'C', 'A'])
keyvalues = rdd.map(lambda x: (x, 1))
counts = keyvalues.reduceByKey(lambda a, b: a + b)
counts.collect()

Out:
[('A', 2), ('B', 1), ('C', 1)]

いじょう。

入門 PySpark ―PythonとJupyterで活用するSpark 2エコシステム

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

PySpark: 時刻と文字列を相互に変換する (DataFrame / Spark SQL)

今回は Apache Spark のインターフェースの一つである PySpark で時刻と文字列を相互に変換する方法について扱う。 PySpark にはいくつかの API があるけど、その中でも DataFrame と Spark SQL を使った方法について紹介する。

使った環境は次の通り。

$ pyspark --version
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.2.1
      /_/
                        
Using Scala version 2.11.8, OpenJDK 64-Bit Server VM, 1.8.0_161
Branch 
Compiled by user felixcheung on 2017-11-24T23:19:45Z
Revision 
Url 
Type --help for more information.

下準備

まずは時刻情報を文字列で表した RDD オブジェクトを用意する。 Apache Spark では主に RDD と DataFrame という二つのオブジェクトを中心にデータを操作する。 その中でも RDD はスキーマレスなオブジェクトになっている。

>>> rdd = sc.parallelize([
...   ('01/Jan/2016 10:13:16',),
...   ('02/Feb/2017 11:14:17',),
...   ('03/Mar/2018 12:15:18',)
... ])

時刻の形式は英語圏にありがちなフォーマットにした。

RDD を DataFrame に変換するためにスキーマ情報を定義する。

>>> from pyspark.sql.types import StructType
>>> from pyspark.sql.types import StructField
>>> from pyspark.sql.types import StringType
>>> schema = StructType([
...   StructField('dt_str', StringType(), False),
... ])

RDD とスキーマを元に DataFrame を生成する。

>>> df = spark.createDataFrame(rdd, schema)
>>> df.show()
+--------------------+
|              dt_str|
+--------------------+
|01/Jan/2016 10:13:16|
|02/Feb/2017 11:14:17|
|03/Mar/2018 12:15:18|
+--------------------+

これで準備ができた。

string から timestamp への変換 (Spark 2.2 ~)

Apache Spark 2.2 以降では、文字列の時刻情報をタイムスタンプに変換する関数として to_timestamp() が用意されている。 この関数は例えば以下のようにして使う。

>>> from pyspark.sql.functions import to_timestamp
>>> df.select(to_timestamp(df.dt_str, 'dd/MMM/yyyy HH:mm:ss').alias('parsed_dt')).show()
+-------------------+
|          parsed_dt|
+-------------------+
|2016-01-01 10:13:16|
|2017-02-02 11:14:17|
|2018-03-03 12:15:18|
+-------------------+

変換した DataFrame の型を調べると、ちゃんと timestamp 型になっていることが分かる。

>>> df.select(to_timestamp(df.dt_str, 'dd/MMM/yyyy HH:mm:ss').alias('parsed_dt')).dtypes
[('parsed_dt', 'timestamp')]

string から timestamp への変換 (Spark 1.5 ~)

先ほど使った API はかなり新しいので使えない環境もあるかと思う。 そこで Apache Spark 1.5 以降であれば次のように unix_timestamp() 関数と cast() メソッドを組み合わせると良い。 ようするに、一旦 UNIX タイムにした上でそれを timestamp として解釈させるということ。

>>> from pyspark.sql.functions import unix_timestamp
>>> df.select(unix_timestamp(df.dt_str, 'dd/MMM/yyyy HH:mm:ss').cast('timestamp').alias('parsed_dt')).show()
+-------------------+
|          parsed_dt|
+-------------------+
|2016-01-01 10:13:16|
|2017-02-02 11:14:17|
|2018-03-03 12:15:18|
+-------------------+

変換した後の DataFrame の型を調べると、ちゃんと timestamp 型が使われていることが分かる。

>>> df.select(unix_timestamp(df.dt_str, 'dd/MMM/yyyy HH:mm:ss').cast('timestamp').alias('parsed_dt')).dtypes
[('parsed_dt', 'timestamp')]

timestamp 型に変換すると、次のように filter() メソッドで範囲指定なんかができる。

>>> parsed_df = df.select(to_timestamp(df.dt_str, 'dd/MMM/yyyy HH:mm:ss').alias('parsed_dt'))
>>> parsed_df.filter('parsed_dt > "2017"').show()
+-------------------+
|          parsed_dt|
+-------------------+
|2017-02-02 11:14:17|
|2018-03-03 12:15:18|
+-------------------+

string から timestamp への変換 (Spark SQL)

ちなみに上記の変換は DataFrame API を使う以外に Spark SQL を使うこともできる。 Spark SQL を使うと、文字通り SQL を通して各種データを操作できる。

Spark SQL を使うには、まず DataFrame を registerTempTable() メソッドを使ってテーブルとして扱えるようにする。

>>> df.registerTempTable('datetimes')

すると SparkSession オブジェクトの sql() メソッドで、上記で登録したテーブルを SQL から触れるようになる。

>>> spark.sql('''
... SELECT
...   CAST(from_unixtime(unix_timestamp(dt_str, 'dd/MMM/yyyy HH:mm:ss')) AS timestamp) AS parsed_dt
... FROM datetimes
... ''').show()
+-------------------+
|          parsed_dt|
+-------------------+
|2016-01-01 10:13:16|
|2017-02-02 11:14:17|
|2018-03-03 12:15:18|
+-------------------+

timestamp を string に変換する (Spark 1.5 ~)

続いては、これまでとは逆にタイムスタンプを文字列に変換してみよう。

タイムスタンプから文字列の変換には Apache Spark 1.5 以降であれば date_format() 関数が使える。

>>> from pyspark.sql.functions import date_format
>>> parsed_df.select(date_format(parsed_df.parsed_dt, 'dd/MMM/yyyy HH:mm:ss').alias('dt_str')).show()
+--------------------+
|              dt_str|
+--------------------+
|01/Jan/2016 10:13:16|
|02/Feb/2017 11:14:17|
|03/Mar/2018 12:15:18|
+--------------------+

変換後の型情報を確認すると、ちゃんと文字列になっていることが分かる。

>>> parsed_df.select(date_format(parsed_df.parsed_dt, 'dd/MMM/yyyy HH:mm:ss').alias('dt_str')).dtypes
[('dt_str', 'string')]

ちなみに、凝ったフォーマットが不要であれば以下のように文字列にキャストしてしまうだけでも事足りる。

>>> parsed_df.select(parsed_df.parsed_dt.cast('string').alias('dt_str')).show()
+-------------------+
|             dt_str|
+-------------------+
|2016-01-01 10:13:16|
|2017-02-02 11:14:17|
|2018-03-03 12:15:18|
+-------------------+

この場合でも、ちゃんと文字列に変換されている。

>>> parsed_df.select(parsed_df.parsed_dt.cast('string').alias('dt_str')).dtypes
[('dt_str', 'string')]

timestamp を string に変換する (Spark SQL)

上記の操作は、もちろん Spark SQL を使ってもできる。 先ほどと同じように、まずは DataFrame を registerTempTable() メソッドでテーブルとして扱えるようにする。

>>> parsed_df.registerTempTable('datetimes')

あとは SparkSession オブジェクト経由で、型を変換する SELECT 文を書けばいいだけ。

>>> spark.sql('''
... SELECT
...   date_format(parsed_dt, 'dd/MMM/yyyy HH:mm:ss') AS dt_str
... FROM datetimes
... ''').show()
+--------------------+
|              dt_str|
+--------------------+
|01/Jan/2016 10:13:16|
|02/Feb/2017 11:14:17|
|03/Mar/2018 12:15:18|
+--------------------+

いじょう。

めでたしめでたし。

入門 PySpark ―PythonとJupyterで活用するSpark 2エコシステム

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

Apache Hive を使ったテーブルのサンプリング

Apache Hive では、大規模なデータセットに対してクエリを実行すると完了までに長い時間がかかる。 そこで、全体から一部を抽出した標本に対してクエリを実行する場合がある。 今回は、その標本を抽出する方法 (サンプリング) について扱う。

使った環境は次の通り。

$ cat /etc/redhat-release 
CentOS Linux release 7.4.1708 (Core)
$ uname -r
3.10.0-693.5.2.el7.x86_64
$ 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 --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

下準備

まずは例となるクエリを実行するテーブルを用意しておこう。 これのテーブルは、整数を格納するカラムを一つだけ持っている。

hive> CREATE TABLE numbers (
    >   n INTEGER
    > );
OK
Time taken: 0.066 seconds

上記のテーブルに対してレコードを追加していく。 ここで注意すべきなのは INSERT 文を個別に発行すること。 詳しくは後述するものの、クエリを一つのまとめてしまうとブロックサンプリングという方法を使ったときに上手く動作しない。

hive> INSERT INTO TABLE numbers VALUES (0);
...
hive> INSERT INTO TABLE numbers VALUES (1);
...
hive> INSERT INTO TABLE numbers VALUES (9);
...
OK
_col0
Time taken: 2.384 seconds

テーブルが以下のような状況になっていることを確認する。

hive> SELECT * FROM numbers;;
OK
numbers.n
0
1
2
3
4
5
6
7
8
9
Time taken: 0.139 seconds, Fetched: 10 row(s)

これで、ひとまず準備ができた。

ランダムなサンプリング

Apache Hive でテーブルの一部をサンプリングするには TABLESAMPLE という構文を使う。 この構文にはいくつかの使い方があるものの、基本は次のようなクエリとなる。

hive> SELECT *
    > from numbers
    > TABLESAMPLE(BUCKET 1 OUT OF 10 ON rand());
OK
numbers.n
6
Time taken: 0.048 seconds, Fetched: 1 row(s)

上記のクエリでは、まず格納されているそれぞれのレコードに rand() 関数でランダムな値を割り振っている。 そして、そのランダムな値をハッシュ化して、結果を 10 個のバケットに振り分けていく。 振り分けられたバケットのうち 1 番目を出力する、というのが上記のクエリの意味となる。

乱数をハッシュ化して 10 個のバケットに割り振っているため、それぞれのバケットには概ね 1 つずつレコードが入ることが期待される。 しかし、もちろん偏ることもあるので次のように 2 つ以上入っていたり、反対に全く入らないこともある。

hive> SELECT *
    > from numbers
    > TABLESAMPLE(BUCKET 1 OUT OF 10 ON rand());
OK
numbers.n
5
6
Time taken: 0.047 seconds, Fetched: 2 row(s)

これはもちろんバケットの数を減らしたり増やした場合にも同じことがいえる。

hive> SELECT *
    > from numbers
    > TABLESAMPLE(BUCKET 1 OUT OF 5 ON rand());
OK
numbers.n
3
6
Time taken: 0.046 seconds, Fetched: 2 row(s)
hive> SELECT *
    > from numbers
    > TABLESAMPLE(BUCKET 1 OUT OF 5 ON rand());
OK
numbers.n
2
5
6
Time taken: 0.062 seconds, Fetched: 3 row(s)

特定のカラムをハッシュ化に用いる

先ほどの例ではハッシュ化に用いる値に rand() 関数が返すランダムな値を使った。 しかし、これにはテーブルに存在する特定のカラムを用いることもできる。

例えば numbers テーブルの n カラムをハッシュ化に使ってみよう。

hive> SELECT *
    > from numbers
    > TABLESAMPLE(BUCKET 1 OUT OF 10 ON n);
OK
numbers.n
0
Time taken: 0.065 seconds, Fetched: 1 row(s)

特定のカラムをハッシュ化に使う場合、値は実行ごとに変化することがないため毎回同じ内容が得られる。

hive> SELECT *
    > from numbers
    > TABLESAMPLE(BUCKET 1 OUT OF 10 ON n);
OK
numbers.n
0
Time taken: 0.068 seconds, Fetched: 1 row(s)

取得したい内容を変更するには、選択するバケットを変えるしかない。

hive> SELECT *
    > from numbers
    > TABLESAMPLE(BUCKET 2 OUT OF 10 ON n);
OK
numbers.n
1
Time taken: 0.057 seconds, Fetched: 1 row(s)
hive> SELECT *
    > from numbers
    > TABLESAMPLE(BUCKET 3 OUT OF 10 ON n);
OK
numbers.n
2
Time taken: 0.053 seconds, Fetched: 1 row(s)

ブロックサンプリング

TABLESAMPLE には特定の割合をサンプリングするよう指定する方法もある。 これはブロックサンプリングと呼ばれるやり方で、その名の通りテーブルを構成するブロック単位でサンプリングする。 下準備で INSERT を一つのクエリにまとめなかったのはこのためだった。 一つのクエリにまとめてしまうと、レコードが全て一つのブロックに格納されてしまうため、この機能が上手く動作しない。

ブロックサンプリングでは TABLESAMPLE に百分率で割合を指定する。

hive> SELECT *
    > FROM numbers
    > TABLESAMPLE(10.0 PERCENT);
OK
numbers.n
0
Time taken: 0.026 seconds, Fetched: 1 row(s)
hive> SELECT *
    > FROM numbers
    > TABLESAMPLE(20.0 PERCENT);
OK
numbers.n
0
1
Time taken: 0.029 seconds, Fetched: 2 row(s)

ただし、この実行結果についても、そのままでは毎回同じ内容が得られる。

hive> SELECT *
    > FROM numbers
    > TABLESAMPLE(20.0 PERCENT);
OK
numbers.n
0
1
Time taken: 0.037 seconds, Fetched: 2 row(s)

得られる内容を変更したいときは、明示的に hive.sample.seednumber を変更してやる必要がある。

hive> set hive.sample.seednumber=7;
hive> SELECT *
    > FROM numbers
    > TABLESAMPLE(20.0 PERCENT);
OK
numbers.n
7
8
Time taken: 0.031 seconds, Fetched: 2 row(s)

まとめ

• Apache Hive でサンプリングするときは TABLESAMPLE を使う
• サンプリングの挙動について
  • 特定の値をハッシュ化してバケットに振り分ける
  • 振り分けられたバケットを選択する
  • ハッシュ化に使う値は rand() 関数の値や、特定のカラムの内容が使える
• 上記のやり方の他にブロックサンプリングという方法もある
  • ブロック単位でサンプリングされる点に注意が必要となる
  • サンプリング結果を変えたいときは hive.sample.seednumber を変更する

プログラミング Hive

情報処理安全確保支援士試験に合格した

お仕事でセキュリティ関連のデータセットも扱う機会が出てきたので、勉強がてら受けてみた。 今回は、受験に関する諸々について書き留めておく。

情報処理安全確保支援士試験について

情報処理安全確保支援士試験というのは、旧情報セキュリティスペシャリスト試験の名前が変わったもの。 IPA が主催している点や、出題範囲やスキルレベルについては旧制度を引き継いでいる。

www.ipa.go.jp

旧制度とは、試験に合格すると情報処理安全確保支援士という士業に登録できる点が異なる。 逆を言えば、登録しない限り「情報処理安全確保支援士」は名乗ることができない。 登録していない場合、書類などには「情報処理安全確保支援士試験合格」とか書くことになる。

支援士に登録したときのメリットは「情報処理安全確保支援士」という名前が名乗れることだけ。 士業といっても、特に独占業務が遂行できるといったものはない。 代わりに、支援士の名称を維持するには毎年お金を払って講習を受ける必要がある。 金額としては、まず登録するときに 1 万円かかるのと、その後の維持費が 3 年間で 15 万円かかるらしい。

講習についても、今のところその金額に見合ったものではないようだ。 なので、とりあえず今の段階では登録はしないつもりでいる。 ちなみに、試験に合格さえしていれば登録自体はいつでもできるようになっている。

試験内容について

試験内容は午前 I, II と午後 I, II に分かれている。 全ての内容で基準点 (60 / 100 点満点中) 以上を取ると合格になる。 今回のスコアはこんな感じだった。

午前 I, II は四択の選択問題で、午後 I, II は一部選択の記述式になっている。 午前 I については別のスキルレベル 4 (ネットワークスペシャリストなど) の試験と共通で、午前 II 以降が独自の問題となる。

午前 I については、別のスキルレベル 4 の試験で基準点以上を 2 年以内に取っていると、申請すれば免除されるらしい。 今回のケースでは、IPA の試験自体が学生以来 10 年ぶりの受験だったので関係なかった。

勉強方法について

勉強方法については、ひとまず一通りの知識を確認しておくために次の本を読んだ。 知らなかったり理解が浅いところについては改めて調べていく感じ。 それ以外は、ほとんど過去問を解くことに時間を費やした。

平成29年度【春期】【秋期】情報処理安全確保支援士 合格教本

午前問題については Web やアプリで空き時間にひたすら反復するのが良いと思う。 似たような問題が例年出るので、基本的に暗記するだけで通る。

情報処理安全確保支援士過去問道場｜情報処理安全確保支援士.com

午後問題については長文を読み解いて回答する必要があるので、こちらは暗記だけでは難しい感じ。 過去問をプリントアウトして何年か分を解いてみることで対策した。

IPA 独立行政法人 情報処理推進機構：過去問題

午後問題で印象的だったのは、設問に「具体的に」と書かれていない限りは概念さえ答えられれば良いということ。 書かれていなければ「これをこうすれば良い (実際にどうやるかは知らん)」というスタンスでも正解になる。 これに気づくまで「もっとちゃんと説明しないと・・・」と回答方法に悩むことが多かった。

受験した感想

学生時代に IPA の試験を受験したときは、午後問題がひたすら難しいと感じていた覚えがある。 今回は、特にそういった印象は受けなかった。 ここらへんは、実務の経験の有無が影響してくるのかもしれない。 ただ、午後問題は一つのトピックについて深掘りしていく感じなので、問題との相性はあるんだろうな。

Apache Hive の MAP 型を試す

前回に引き続き Apache Hive の複合型の一つ MAP 型を試してみる。

blog.amedama.jp

MAP 型は一般的なプログラミング言語でいうマップや辞書といったデータ構造に相当する。 これを使うとテーブルのカラムに任意のキーで値を格納できる。

環境は次の通り。

$ cat /etc/redhat-release 
CentOS Linux release 7.4.1708 (Core)
$ uname -r
3.10.0-693.5.2.el7.x86_64
$ 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 --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

MAP 型を使ったテーブルを作る

MAP 型を使ってテーブルを作るときは、次のようにキーと値に使う型を指定してカラムを作る。

hive> CREATE TABLE users (
    >   name STRING,
    >   property MAP<STRING, STRING>
    > );
OK
Time taken: 0.047 seconds

ちなみにキーにはプリミティブ型しか指定できない。 試しにキーに ARRAY 型を指定してみると、以下のようにエラーになる。

hive> CREATE TABLE example (
    >   test MAP<ARRAY, STRING>
    > );
...
FAILED: ParseException line 2:11 cannot recognize input near 'ARRAY' ',' 'STRING' in primitive type specification

MAP 型を含むレコードを追加する

他の複合型と同様に、一般的な INSERT INTO ... VALUES を使ったレコードの追加ができない。

hive> INSERT INTO TABLE users
    >   VALUES (MAP("key1", "value1", "key2", "value2"));
FAILED: SemanticException [Error 10293]: Unable to create temp file for insert values Expression of type TOK_FUNCTION not supported in insert/values

代わりに SELECT を使ってデータを作って、それを追加してやる。 MAP() 関数を使うことでキーと値の組を使ったカラムを作れる。

hive> SELECT MAP("key1", "value1", "key2", "value2");
OK
_c0
{"key1":"value1","key2":"value2"}
Time taken: 0.059 seconds, Fetched: 1 row(s)

上記の SELECT で作ったデータを INSERT INTO でテーブルに追加する。

hive> INSERT INTO TABLE users
    >   SELECT "Alice", MAP("key1", "value1", "key2", "value2");
...
OK
_c0 _c1
Time taken: 19.437 seconds

テーブルを確認すると、ちゃんとレコードが入っている。

hive> SELECT * FROM users;
OK
users.name  users.property
Alice   {"key1":"value1","key2":"value2"}
Time taken: 0.136 seconds, Fetched: 1 row(s)

MAP 型の中身を参照する

MAP 型の中身を取り出すときはブラケットの中にキーを指定する。

hive> SELECT name, property["key1"] FROM users;
OK
name    _c1
Alice   value1
Time taken: 0.157 seconds, Fetched: 1 row(s)

存在しないキーの振る舞いを確認する

MAP 型には型さえ合っていれば、これまでに追加したことのないキーであっても大丈夫。

hive> INSERT INTO TABLE users
    >   SELECT "Bob", MAP("key1", "value1", "key3", "value3");
...
OK
_c0 _c1
Time taken: 18.935 seconds

テーブルの内容を確認すると、レコードごとに別々のキーが入っていることが分かる。

hive> SELECT * FROM users;
OK
users.name  users.property
Alice   {"key1":"value1","key2":"value2"}
Bob {"key1":"value1","key3":"value3"}
Time taken: 0.133 seconds, Fetched: 2 row(s)

レコードによってあったりなかったりするキーを参照したときの振る舞いを確認しておこう。

hive> SELECT name,
    >        property["key1"] AS key1,
    >        property["key2"] AS key2,
    >        property["key3"] AS key3
    > FROM users;
OK
name    key1    key2    key3
Alice   value1  value2  NULL
Bob value1  NULL    value3
Time taken: 0.123 seconds, Fetched: 2 row(s)

上記のように、レコードにキーがないときは NULL になるようだ。

次の確認のために、一旦テーブルを削除しておこう。

hive> DROP TABLE users;
OK
Time taken: 0.073 seconds

外部ファイルからデータを読み込む

続いて、外部ファイルからデータを読み込むときの挙動も確認しておく。

外部ファイルから読み込むときは次のように各フィールドやカラムの分割文字を指定しておく。

hive> CREATE TABLE users (
    >   name STRING,
    >   property MAP<STRING, STRING>
    > )
    > ROW FORMAT DELIMITED 
    > FIELDS TERMINATED BY ','  
    > COLLECTION ITEMS TERMINATED BY '$'
    > MAP KEYS TERMINATED BY ':'
    > STORED AS TEXTFILE;
OK
Time taken: 0.075 seconds

次のように CSV ファイルを作っておく。 先のテーブルを定義するときに指定した通り、キーと値は : で区切って、フィールドは $ で分割する。

$ cat << 'EOF' > users.csv
Alice,key1:value1$key2:value2
Bob,key1:value1$key3:value3
EOF

上記を Hive で LOAD DATA を使って読み込む。

hive> LOAD DATA LOCAL INPATH '/home/vagrant/users.csv' INTO TABLE users;
Loading data to table default.users
OK
Time taken: 0.385 seconds

すると、次のようにちゃんとテーブルにデータが格納された。

hive> SELECT * FROM users;
OK
users.name  users.property
Alice   {"key1":"value1","key2":"value2"}
Bob {"key1":"value1","key3":"value3"}
Time taken: 0.13 seconds, Fetched: 2 row(s)

ばっちり。

プログラミング Hive

Apache Hive の STRUCT 型を試す

Apache Hive には基本となる文字列や数値以外にも複合型 (Complex Type) というデータタイプがある。 以前、その中の一つとして ARRAY 型をこのブログでも扱った。

blog.amedama.jp

今回は、それに続いて複合型の中で STRUCT 型というデータタイプを試してみる。 これは、文字通り一般的なプログラミング言語でいう構造体 (Struct) に相当するもの。 この STRUCT 型を使うことで一つのカラムの中に複数のデータを格納できる。 使い勝手としては KVS によくあるカラムファミリーに近いかもしれない。

環境は次の通り。

$ cat /etc/redhat-release 
CentOS Linux release 7.4.1708 (Core)
$ uname -r
3.10.0-693.5.2.el7.x86_64
$ 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 --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

STRUCT 型を使ってテーブルを作る

例として、まずは名前の姓と名を STRUCT 型で分割して保存するテーブルを作ってみる。 STRUCT 型を定義するときは、次のように <> の中に名前と型をカンマ区切りで羅列していく。

hive> CREATE TABLE users (
    >   name STRUCT<first: STRING,
    >               last: STRING>
    > );
OK
Time taken: 0.135 seconds

STRUCT 型を使ったレコードを追加する

データを追加するときは一般的な INSERT INTO ... VALUES が使えない。

hive> INSERT INTO TABLE users
    >   VALUES (NAMED_STRUCT("first", "John", "last", "Doe"));
FAILED: SemanticException [Error 10293]: Unable to create temp file for insert values Expression of type TOK_FUNCTION not supported in insert/values

代わりに SELECT で作ったデータを使うことになる。 NAMED_STRUCT() 関数はフィールドに名前のついた STRUCT 型のデータを作るために使う。

hive> SELECT NAMED_STRUCT("first", "John", "last", "Doe");
OK
_c0
{"first":"John","last":"Doe"}
Time taken: 0.066 seconds, Fetched: 1 row(s)

上記のように SELECT で作ったデータを INSERT INTO に渡してやる。

hive> INSERT INTO TABLE users
    >   SELECT NAMED_STRUCT("first", "John", "last", "Doe");
...
OK
_c0
Time taken: 21.451 seconds

この通り、ちゃんとレコードが保存された。

hive> SELECT * FROM users;
OK
users.name
{"first":"John","last":"Doe"}
Time taken: 0.156 seconds, Fetched: 1 row(s)

STRUCT 型のフィールドを参照する

STRUCT 型に保存されたフィールドの中身を参照するときは、次のようにカラム名にドットでフィールド名をつなげてやる。

hive> SELECT name.first, name.last FROM users;
OK
first   last
John    Doe
Time taken: 0.143 seconds, Fetched: 1 row(s)

基本的な使い方は上記の通り。 一旦テーブルを削除しておこう。

hive> DROP TABLE users;
OK
Time taken: 0.119 seconds

ARRAY 型と組み合わせて使う

STRUCT 型は別の複合型と組み合わせて使うこともできる。

例えば ARRAY 型の中に STRUCT 型を含むようなテーブルを作ってみよう。

hive> CREATE TABLE users (
    >   name STRING,
    >   addresses ARRAY<STRUCT<country: STRING,
    >                          city: STRING>>
    > );
OK
Time taken: 0.114 seconds

データを追加するときは、次のように ARRAY() 関数と NAMED_STRUCT() 関数を組み合わせる。

hive> INSERT INTO TABLE users
    >   SELECT "Alice", ARRAY(NAMED_STRUCT("country", "japan", "city", "tokyo"),
    >                         NAMED_STRUCT("country", "japan", "city", "osaka"));
...
OK
_c0 _c1
Time taken: 20.376 seconds

テーブルを確認すると、ちゃんと ARRAY 型の中に STRUCT 型のデータが収まっていることが分かる。

hive> SELECT * FROM users;
OK
users.name  users.addresses
Alice   [{"country":"japan","city":"tokyo"},{"country":"japan","city":"osaka"}]
Time taken: 0.133 seconds, Fetched: 1 row(s)

中身を展開して集計するときは普通に ARRAY 型を使うときと同じように LATERAL VIEW と explode() 関数を組み合わせれば良い。

hive> SELECT *
    > FROM users
    > LATERAL VIEW explode(addresses) users AS address;
OK
users.name  users.addresses users.address
Alice   [{"country":"japan","city":"tokyo"},{"country":"japan","city":"osaka"}] {"country":"japan","city":"tokyo"}
Alice   [{"country":"japan","city":"tokyo"},{"country":"japan","city":"osaka"}] {"country":"japan","city":"osaka"}
Time taken: 0.121 seconds, Fetched: 2 row(s)
hive> SELECT name, address.country, address.city
    > FROM users
    > LATERAL VIEW explode(addresses) users AS address;
OK
name    country city
Alice   japan   tokyo
Alice   japan   osaka
Time taken: 0.045 seconds, Fetched: 2 row(s)

外部ファイルからデータを読み込む

外部ファイルからデータを読み込むときは、次のようにフィールドやコレクションの区切り文字を指定しておく。

hive> CREATE TABLE users (
    >   name STRUCT<first: STRING,
    >               last: STRING>
    > )
    > ROW FORMAT DELIMITED 
    > FIELDS TERMINATED BY ','  
    > COLLECTION ITEMS TERMINATED BY '$'
    > STORED AS TEXTFILE;
OK
Time taken: 0.143 seconds

フィールドの区切り文字として $ を使った CSV ファイルを用意しておく。

$ cat << 'EOF' > users.csv 
Yamada$Taro
Suzuki$Ichiro
EOF

あとは上記を Hive で読み込むだけ。

hive> LOAD DATA LOCAL INPATH '/home/vagrant/users.csv' INTO TABLE users;
Loading data to table default.users
OK
Time taken: 0.944 seconds

この通り、ちゃんとデータが格納された。

hive> SELECT * FROM users;
OK
users.name
{"first":"Yamada","last":"Taro"}
{"first":"Suzuki","last":"Ichiro"}
Time taken: 0.287 seconds, Fetched: 2 row(s)

次はもうちょっと複雑な例を示す。 その前に、一旦テーブルを削除しておこう。

hive> DROP TABLE users;
OK
Time taken: 0.193 seconds

次は、先ほどと同じように ARRAY 型と STRUCT 型を組み合わせたパターンでも外部ファイルから読み込んでみる。 このときのポイントとしては MAP KEYS TERMINATED BY も指定しておくところ。

hive> CREATE TABLE users (
    >   name STRING,
    >   addresses ARRAY<STRUCT<country: STRING,
    >                          city: STRING>>
    > )
    > ROW FORMAT DELIMITED 
    > FIELDS TERMINATED BY ','  
    > COLLECTION ITEMS TERMINATED BY '$'
    > MAP KEYS TERMINATED BY ':'
    > STORED AS TEXTFILE;
OK
Time taken: 0.06 seconds

今度はフィールドの区切り文字は : を使いつつリストの区切り文字として $ を指定してやる。

$ cat << 'EOF' > users.csv
Alice,japan:tokyo$japan:osaka
Bob,america:newyork$america:california
EOF

上記のファイルを読み込んでみよう。

hive> LOAD DATA LOCAL INPATH '/home/vagrant/users.csv' INTO TABLE users;
Loading data to table default.users
OK
Time taken: 0.432 seconds

すると、以下のようにちゃんと保存されている。

hive> SELECT * FROM users;
OK
users.name  users.addresses
Alice   [{"country":"japan","city":"tokyo"},{"country":"japan","city":"osaka"}]
Bob [{"country":"america","city":"newyork"},{"country":"america","city":"california"}]
Time taken: 0.134 seconds, Fetched: 2 row(s)

ばっちり。

プログラミング Hive

Apache Hive の Partition 機能を使ってみる

今回は Apache Hive の Partition 機能を使ってみる。 Partition 機能を用いない場合、クエリを発行するとテーブルを構成するファイル群にフルスキャンがかかる。 それに対し、Partition 機能を用いるとクエリによってはスキャンするファイルの範囲を制限できる。 結果としてパフォーマンスの向上が見込める場合がある。

使った環境は次の通り。 Apache Hive や Hadoop のインストール部分については省略する。

$ cat /etc/redhat-release 
CentOS Linux release 7.4.1708 (Core)
$ uname -r
3.10.0-693.5.2.el7.x86_64
$ 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 --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

下準備として Hive の CLI を起動しておく。

$ hive
hive> set hive.cli.print.header=true;

Partition 機能を使わない場合

まずは Partition 機能を使わない場合について見ておこう。

例としてユーザがログインしたときのイベントを記録するテーブルを用意する。

hive> CREATE TABLE login_events (
    >   datetime TIMESTAMP,
    >   name STRING
    > );
OK
Time taken: 0.052 seconds

上記のテーブルにレコードを追加してみよう。

hive> INSERT INTO login_events
    >   VALUES ("2018-01-01 10:00:00", "Alice");
...
OK
_col0   _col1
Time taken: 22.584 seconds

一件のレコードがテーブルに保存された。

hive> SELECT * FROM login_events;
OK
login_events.datetime   login_events.name
2018-01-01 10:00:00 Alice
Time taken: 0.148 seconds, Fetched: 1 row(s)

このとき、実際にテーブルのデータが格納される場所は SHOW CREATE TABLE を使って確認できる。 具体的には LOCATION のところ。

hive> SHOW CREATE TABLE login_events;
OK
createtab_stmt
CREATE TABLE `login_events`(
  `datetime` timestamp, 
  `name` string)
ROW FORMAT SERDE 
  'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe' 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.mapred.TextInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION
  'hdfs://master:9000/user/hive/warehouse/login_events'
TBLPROPERTIES (
  'transient_lastDdlTime'='1514916341')
Time taken: 0.026 seconds, Fetched: 13 row(s)

今回であれば HDFS 上の /user/hive/warehouse/login_events というパスに格納されていることが分かる。

hdfs コマンドを使って上記のディレクトリを確認してみよう。 たしかに、先ほど追加したデータが書き込まれたテキストファイルがある。

$ hdfs dfs -ls /user/hive/warehouse/login_events
Found 1 items
-rwxrwxr-x   2 vagrant supergroup         26 2018-01-02 18:26 /user/hive/warehouse/login_events/000000_0
$ hdfs dfs -cat /user/hive/warehouse/login_events/000000_0
2018-01-01 10:00:00Alice

さて、上記のテーブルは Partition 機能を使っていない。 そのため、以下のような時刻を制限するようなクエリを実行したときも、ファイル群のフルスキャンが必要になってしまう。

hive> SELECT *
    >   FROM login_events
    >   WHERE datetime > "2018-01-01 00:00:00";
OK
login_events.datetime   login_events.name
2018-01-01 10:00:00 Alice
Time taken: 0.169 seconds, Fetched: 1 row(s)

フルスキャンが走るとデータが増えたときに実行時間がかかるので、それを避けたいというのが今回の本題になる。

一旦、上記のテーブルを削除しておこう。

hive> DROP TABLE login_events;
OK
Time taken: 0.143 seconds

Table-by-Day アンチパターン

Partition 機能の説明に入る前に Table-by-Day というアンチパターンについて見ておこう。 これは、フルスキャンに時間のかかるテーブルに対処する方法として使われることのあるやり方だ。 具体的には、日付ごとの単位などでテーブルを分割して作ることでスキャンの範囲を狭めようというもの。

hive> CREATE TABLE login_events_20180101 (
    >   datetime TIMESTAMP,
    >   name STRING
    > );
OK
Time taken: 0.107 seconds
hive> CREATE TABLE login_events_20180102 (
    >   datetime TIMESTAMP,
    >   name STRING
    > );
OK
Time taken: 0.046 seconds

もちろんデータはテーブルに応じて格納することになる。

hive> INSERT INTO login_events_20180101
    >   VALUES ("2018-01-01 10:00:00", "Alice");
...
OK
_col0   _col1
Time taken: 23.413 seconds
hive> INSERT INTO login_events_20180102
    >   VALUES ("2018-01-02 20:00:00", "Bob");
...
OK
_col0   _col1
Time taken: 25.033 seconds

ただ、このやり方を取ると日付をまたいだ集計をするときなんかにクエリが複雑化してイマイチな感じ。

hive> SELECT * FROM login_events_20180101
    > UNION ALL
    > SELECT * FROM login_events_20180102;
...
_u1.datetime    _u1.name
2018-01-01 10:00:00 Alice
2018-01-02 20:00:00 Bob
Time taken: 26.36 seconds, Fetched: 2 row(s)

Apache Hive では Table-by-Day よりも Partition 機能を使うことが推奨されている。

Partition 機能を使った場合

続いて Partition 機能を使ってみる。

今度はテーブルを作るときに PARTITIONED BY で年月日 (year, month, day) を元にパーティションを構成する。

hive> CREATE TABLE login_events (
    >   datetime TIMESTAMP,
    >   name STRING
    > )
    > PARTITIONED BY (year INT, month INT, day INT);
OK
Time taken: 0.049 seconds

datetime を直接使えないの？という疑問が浮かぶと思うけど残念ながら難しそう。

レコードを追加するときは、次のようにカラムの値と一緒にパーティションの値も指定することになる。

hive> INSERT INTO login_events
    >   PARTITION (year=2017, month=1, day=1)
    >   VALUES ("2018-01-01 10:00:00", "Alice");
...
_col0   _col1
Time taken: 24.017 second

さて、それでは先ほどと同じように HDFS 上でデータがどのように格納されているかを確認してみよう。

hive> SHOW CREATE TABLE login_events;
OK
createtab_stmt
CREATE TABLE `login_events`(
  `datetime` timestamp, 
  `name` string)
PARTITIONED BY ( 
  `year` int, 
  `month` int, 
  `day` int)
ROW FORMAT SERDE 
  'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe' 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.mapred.TextInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION
  'hdfs://master:9000/user/hive/warehouse/login_events'
TBLPROPERTIES (
  'transient_lastDdlTime'='1514917075')
Time taken: 0.048 seconds, Fetched: 17 row(s)

格納されているディレクトリを見てみると、パーティションの値に応じてディレクトリが作られていることが分かる。

$ hdfs dfs -ls /user/hive/warehouse/login_events
Found 1 items
drwxrwxr-x   - vagrant supergroup          0 2018-01-02 18:18 /user/hive/warehouse/login_events/year=2017
$ hdfs dfs -find /user/hive/warehouse/login_events
/user/hive/warehouse/login_events
/user/hive/warehouse/login_events/year=2017
/user/hive/warehouse/login_events/year=2017/month=1
/user/hive/warehouse/login_events/year=2017/month=1/day=1
/user/hive/warehouse/login_events/year=2017/month=1/day=1/000000_0

そう、つまり Partition 機能というのは格納されるディレクトリを分割することで実現されている。

新しくレコードを追加したときも確認してみよう。 今度はパーティションの値が先ほどとは異なる。

hive> INSERT INTO login_events
    >   PARTITION (year=2017, month=1, day=2)
    >   VALUES ("2018-01-02 20:00:00", "Bob");
...
OK
_col0   _col1
Time taken: 20.819 seconds

同様に、新しくパーティションに対応するディレクトリが作成された。

$ hdfs dfs -find /user/hive/warehouse/login_events
/user/hive/warehouse/login_events
/user/hive/warehouse/login_events/year=2017
/user/hive/warehouse/login_events/year=2017/month=1
/user/hive/warehouse/login_events/year=2017/month=1/day=1
/user/hive/warehouse/login_events/year=2017/month=1/day=1/000000_0
/user/hive/warehouse/login_events/year=2017/month=1/day=2
/user/hive/warehouse/login_events/year=2017/month=1/day=2/000000_0

この状態でパーティションを使って条件式を組むとファイルのスキャン範囲がディレクトリ内に限定されるという寸法らしい。

hive> SELECT *
    > FROM login_events
    > WHERE year = 2017 AND month = 1 AND day = 1;
OK
login_events.datetime   login_events.name   login_events.year   login_events.month  login_events.day
2018-01-01 10:00:00 Alice   2017   1  1
Time taken: 0.389 seconds, Fetched: 1 row(s)

なるほど、それなら条件によってはクエリのパフォーマンスが向上するだろう、という感じ。

一旦、またテーブルを削除しておこう。

hive> DROP TABLE login_events;
OK
Time taken: 0.499 seconds

既存のテーブルにパーティションを追加する

続いては、既存の Partition 機能を使っていないテーブルにパーティションを追加したいというユースケースについて。 ようするに、最初は余裕だったけど時間とともにデータが増えてきてやばっ、というような場合。

まずは Partition 機能を使わない状態でテーブルを作ってレコードを追加しておこう。

hive> CREATE TABLE login_events (
    >   datetime TIMESTAMP,
    >   name STRING
    > );
OK
Time taken: 0.156 seconds
hive> INSERT INTO login_events
    >   VALUES
    >     ("2018-01-01 10:00:00", "Alice"),
    >     ("2018-01-02 20:00:00", "Bob"),
    >     ("2018-01-03 06:00:00", "Carol");
OK
_col0   _col1
Time taken: 22.689 seconds

あらかじめ、ここからの作業の流れについて概要を説明しておく。 残念なことに、既存のテーブルに直接パーティションのカラムを追加することはできない。 そこで、まずは既存のテーブルにパーティションを追加した形で新しいテーブルを用意しておく。 そして、その新しいテーブルに既存のテーブルのデータを移し替えることになる。

Partition に使う値は、元々あった datetime カラムから生成することにしよう。 先ほどはパーティションを year > month > day と階層構造にしたけど、今度は一つにまとめてしまうことにする。

hive> SELECT date_format(datetime, "yyyy-MM-DD") FROM login_events;
OK
_c0
2018-01-01
2018-01-02
2018-01-03
Time taken: 0.253 seconds, Fetched: 3 row(s)

データを移行する先の、パーティションを追加した新しいテーブルを partitioned_login_events という名前で用意する。

hive> CREATE TABLE partitioned_login_events (
    >   datetime TIMESTAMP,
    >   name STRING
    > )
    > PARTITIONED BY (day STRING);
OK
Time taken: 0.137 seconds

これから実施する作業は Dynamic Partition と呼ばれる機能を使うことになる。 そこで、あらかじめその機能を有効にしておこう。

hive> SET hive.exec.dynamic.partition=true;
hive> SET hive.exec.dynamic.partition.mode=nonstrict;

Dynamic Partition 機能というのは、既存のテーブルに存在するカラムの値から自動的にパーティションを構成する機能をいう。 この機能を使わないと、自分で一つ一つパーティションを指定しながらデータの移行作業をする羽目になってつらい。

実際にデータの移行を実施するクエリは次の通り。 login_events から partitioned_login_events に対して INSERT OVERWRITE でデータを移行している。 その際に SELECT で作成した day カラムをパーティションに指定しているのがポイント。

hive> FROM login_events
    > INSERT OVERWRITE TABLE partitioned_login_events
    > PARTITION(day)
    > SELECT datetime, name, date_format(datetime, "yyyy-MM-DD") AS day;
...
OK
datetime    name    day
Time taken: 21.565 seconds

データがどのように格納されたかを確認してみよう。

hive> SHOW CREATE TABLE partitioned_login_events;
OK
createtab_stmt
CREATE TABLE `partitioned_login_events`(
  `datetime` timestamp, 
  `name` string)
PARTITIONED BY ( 
  `day` string)
ROW FORMAT SERDE 
  'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe' 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.mapred.TextInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION
  'hdfs://master:9000/user/hive/warehouse/partitioned_login_events'
TBLPROPERTIES (
  'transient_lastDdlTime'='1514918693')
Time taken: 0.02 seconds, Fetched: 15 row(s)

ちゃんと日付に応じてディレクトリが作られており Partition 機能が有効に働いていることが分かる。

$ hdfs dfs -find /user/hive/warehouse/partitioned_login_events
/user/hive/warehouse/partitioned_login_events
/user/hive/warehouse/partitioned_login_events/day=2018-01-01
/user/hive/warehouse/partitioned_login_events/day=2018-01-01/000000_0
/user/hive/warehouse/partitioned_login_events/day=2018-01-02
/user/hive/warehouse/partitioned_login_events/day=2018-01-02/000000_0
/user/hive/warehouse/partitioned_login_events/day=2018-01-03
/user/hive/warehouse/partitioned_login_events/day=2018-01-03/000000_0

あとは元々の名前にテーブルをリネームするなりご自由に。

まとめ

• Apache Hive は、そのままだとテーブルを構成するファイル群をフルスキャンする
• スキャン範囲を限定する方法には Table-by-Day というアンチパターンがある
• Apache Hive では代わりに Partition 機能を使うことが推奨されている
• Partition 機能はデータを格納するディレクトリを分割することで実現されている

プログラミング Hive