mixi engineer blog

*** 引っ越しました。最新の情報はこちら → https://medium.com/mixi-developers *** ミクシィ・グループで、実際に開発に携わっているエンジニア達が執筆している公式ブログです。様々なサービスの開発や運用を行っていく際に得た技術情報から採用情報まで、有益な情報を幅広く取り扱っています。

今日からはじめるお手軽 Hive データ移行

こんにちは. 昨年知人のオーケストラ演奏会で聴いたメンデルスゾーン交響曲第4番「イタリア」が大好きな,技術部の石川有です.

そんな「イタリア」大好きな私ですが Hive のデータ移行も大好きという体で, 今回の記事ではオンプレミスで Hive を頑張って運用している方たちに向けて,どうしてもというときの Hive のデータ移行方法を紹介したいと思います. 弊社では最近,Cloudera's Distribution including Apache Hadoop 3 update 2 (CDH3u2) から CDH 4.1.2 にバージョンアップを行なっています. このようにオンプレミスで Hive を運用している方たちの中には,すでに持っているデータをうまく移行したいという方も多いかと思います. Hive には現状 MySQL の mysqldump のような機能が提供されていないのでバージョンを上げるために,別の Hadoop クラスタを用意して利用するシステムを移行するときに役立つ方法だと思います.

本文章の構成

まず最初に,弊社における Hadoop のバージョンを上げることにともなう問題について述べます. つぎに,Hadoop クラスタ間で Hive のデータ移行を行うための基礎知識を話します. それから,Hadoop クラスタ間で Hive のデータを移行するための具体的な手順を述べます. またこの仕組みを利用した,さらなる応用方法を述べ,おわりとします. 

  • Hadoop のバージョンを上げるときの問題
  • 基礎知識
  • データ移行をするための手順
  • さらなる応用
  • おわりに

Hadoop のバージョンを上げるときの問題

弊社では Apache Hadoop / Hive を日常的に利用し,定期実行処理も多く実行されています. このように稼働しているシステムを止めるわけには行かないので,まったく別のクラスタを構築し無事稼働するかを見届けて運用を切り替えるというアプローチを取りました. 特に CDH のメジャーバージョンアップは変更点が多く,安定稼働するための検証などを行う必要がありました. (余談ですが,CDH 4.1.1 の時点では 圧縮のデータが読めないバグや CDH 4.1.2 でも Hive の Multiple Insert という機能にバグがあるという問題があります)

Hadoop クラスタの再構築で問題となることのひとつに,格納されているデータを再構築するということだと思います. 通常のログとして保存されているデータであれば,地道にデータをロードするという方法もあるかと思います. しかし,元データがあるものは別として,そこから生み出された中間データの再生成かかるコストの問題があります.

いくら Hadoop が大規模データを扱うことに長けているとはいえ,日々の解析をより短時間にするためにサマライズした中間データが必要になります. DataNode 18 台という決して大規模とは言えないクラスタですら,単純な行数カウントであれば 700 億行超を 1000 秒程度で処理できます. これほどの高機能であっても,すべての中間データを再生成するのはあまりにもコストが高すぎます. たとえば,1日分のアクセスログをサマライズするのに 30 秒かかるとすると,100 日分で 3000 秒かかってしまいます. これが1つだけなら良いですが,何種類もサマライズするデータがあればそれだけ時間がかかってしまいます. そのためデータをクラスタ間で移行する手段があれば,再生成のコストがなくなるので非常に嬉しいです. そのようなときの姑息な手段として,Hive のデータをクラスタ間で移行する方法を紹介します.

基礎知識

Hive のデータの格納先

Hive のデータは,HDFS 上に普通に存在しています. デフォルトの設定であれば,HDFS の /user/hive/warehouse/ 以下で確認することができます. たとえば,hoge HiveDB fuga HiveTABLE という Hive テーブルであれば,/user/hive/warehouse/hoge.db/fuga 以下に格納されます.


hive> CREATE TABLE IF NOT EXISTS (
  column1 STRING,
  column2 STRING
)
PARTITIONED BY (dt STRING, type STRING)
STORED AS RCFILE;

また Hive テーブルに PARTITION といういわゆるデータの「区切りを」定義したときは,PARTITION の key, value が HDFS 上のパスになります. つまり,HDFS のパス情報から格納されている PARTITION の情報が分かります.


/user/hive/warehouse/hoge.db/fuga/dt=2013-01-01/type=A/000000_1
/user/hive/warehouse/hoge.db/fuga/dt=2013-01-01/type=B/000000_1
/user/hive/warehouse/hoge.db/fuga/dt=2013-01-02/type=A/000000_1
/user/hive/warehouse/hoge.db/fuga/dt=2013-01-02/type=B/000000_1
/user/hive/warehouse/hoge.db/fuga/dt=2013-01-02/type=C/000000_1

Hive のデータロード

Hive のデータロードは,Linux などのローカルファイルからロードする経路と HDFS にすでに存在するデータをロードする経路の2つがあります. 前者の Linux などのローカルファイルをロードするときは,データのコピーが行われます. つまり,Linux 上のファイルと同じものが Hive にロードされます. それに対して HDFS にすでに存在するデータを Hive にロードするときは,データが HDFS 内部的に移動されます. つまり,Linux の同一ファイルシステム上での mv をイメージしていただけると良いです. そのため HDFS からの LOAD DATA は,ほとんど負荷なく行えるというメリットがあります. そして,Hive の LOAD DATA コマンドは,元データに対して「なにかしら」の処理を加えず加えません. たとえば LOAD DATA で Hive テーブルにロードすると,全く同じ gzip が HDFS にコピーされていることが確認できます.

簡単なオペレーションの例を,つぎに示します. HDFS 上に Hive にロードしたいデータが,HDFS の /temporary 以下にあるとします. このデータのパスから,どの PARTITION に格納されるべきデータかがわかります. その情報を利用して,LOAD DATA コマンドを実行します. その結果,Hive 側の HDFS のパスに移動することが確認できます.


## 元データの HDFS のパス
/temporary/hoge.db/fuga/dt=2013-01-03/type=C/data.gzip

## Hive テーブルへのデータロード
hive> LOAD DATA INPATH '/temporary/hoge.db/fuga/dt=2013-01-03/type=C/*'
hive> PARTITION (dt = '2013-01-03', type = 'C')
hive> OVERWRITE INTO TABLE `hoge`.`fuga`;

## ロードされたデータ
/user/hive/warehouse/hoge.db/fuga/dt=2013-01-03/type=C/data.gzip

load_data_flow.jpg

データ移行をするための手順

Hive のデータをクラスタ間で移行する手順は,大きく分けると3つの作業になります. まず,移動したいデータを distcp を利用してクラスタ間でデータ転送します. つぎに,再構築したい Hive テーブルを移行先のクラスタに作ります. Hive テーブルを作っておかないと,当然ながら LOAD DATA が実行できないので忘れないようにしてください. 転送されたデータの HDFS のパス情報からロード先の PARTITION を特定して,スクリプトなどを利用してひたすら LOAD DATA を実行します.

  1. 移動したいデータを distcp でクラスタ間で転送
  2. 再構築したい Hive テーブルのスキーマを作成
  3. ひたすら HDFS のデータから Hive テーブルにロード

move_data_inter_hadoop_cluster.jpg

クラスタ間でデータ転送

Hadoop には,distcp という Hadoop クラスタ間でデータ転送をするツールがあります. このツールを利用して,古いクラスタから新しいクラスタにデータ転送を行います. CDH4 と CDH3 では HDFS のプロトコルがことなるため,hftp を利用します. hdfs のポートはデフォルトでは 8020 で,hftp のポートは 50070 です. 今回は敢えて,新しい方のクラスタの HDFS のパスを /temporary/ 以下として,Hive が格納される HDFS のパスとは分けています.


## 新しいクラスタで実行
## 新しいクラスタの HDFS に /temporary/hoge.db/fuga/ が転送される
$> hadoop distcp -i -m 5 hftp://old-cluster:50070/user/hive/warehouse/hoge.db/fuga/ hdfs://new-cluster:8020/temporary/hoge.db/fuga/

再構築したい Hive テーブルのスキーマを作成

前述の例を再利用しますが,再構築したい Hive テーブルを新しいクラスタにも作ります. Hive の CREATE TABLE を新クラスタでも実行します.


CREATE TABLE IF NOT EXISTS (
  column1 STRING,
  column2 STRING
)
PARTITIONED BY (dt STRING, type STRING)
STORED AS RCFILE;

ひたすら HDFS のデータから Hive テーブルにロード

転送した結果,つぎのように HDFS 上にデータがあるとします. hoge HiveDB fuga HiveTABLE のデータの各 PARTITION が HDFS のパスで別れて存在することになります.


/temporary/hoge.db/fuga/dt=2013-01-01/type=A/000000_1
/temporary/hoge.db/fuga/dt=2013-01-01/type=B/000000_1
/temporary/hoge.db/fuga/dt=2013-01-02/type=A/000000_1
/temporary/hoge.db/fuga/dt=2013-01-02/type=B/000000_1

この HDFS のパス名からどの PARTITION にどのパスのデータを格納すればいいのかわかるので,スクリプトでも書いて簡単に LOAD DATA コマンドを生成できます. たとえば,各パスのデータをデータロードする Hive のコマンドは,つぎのようになります. このような方法であれば,クラスタ間の Hive のデータ移行を無理やり行うことができました.


hive> LOAD DATA INPATH '/temporary/hoge.db/fuga/dt=2013-01-01/type=A/*'
hive> PARTITION (dt = '2013-01-01', type = 'A')
hive> OVERWRITE INTO TABLE `hoge`.`fuga`;

hive> LOAD DATA INPATH '/temporary/hoge.db/fuga/dt=2013-01-01/type=B/*'
hive> PARTITION (dt = '2013-01-01', type = 'B')
hive> OVERWRITE INTO TABLE `hoge`.`fuga`;

hive> LOAD DATA INPATH '/temporary/hoge.db/fuga/dt=2013-01-02/type=A/*'
hive> PARTITION (dt = '2013-01-02', type = 'A')
hive> OVERWRITE INTO TABLE `hoge`.`fuga`;

hive> LOAD DATA INPATH '/temporary/hoge.db/fuga/dt=2013-01-02/type=B/*'
hive> PARTITION (dt = '2013-01-02', type = 'A')
hive> OVERWRITE INTO TABLE `hoge`.`fuga`;

さらなる応用

またクラスタ間のデータ移行ではないにしても,Hive テーブルの定義を間違えたようなときにも利用できます. たとえば,Hive テーブル定義に正規表現がつかえる org.apache.hadoop.hive.contrib.serde2.RegexSerDe があります. これはアクセスログなどの半構造化データを,正規表現で文字列の要素を Hive テーブルのカラムとして分解することができます.

org.apache.hadoop.hive.contrib.serde2.RegexSerDe の問題として,ALTER TABLE で後から正規表現を変更して,Hive テーブルのカラムを追加したりすることができません. つまり,一度定義したら変更を加えるのにデータを再ロードする必要があります. このようなとき,上記で述べたような方法で「無理やり」データを再構築できます.
この方法のメリットは,「Hive のデータロード」で述べた HDFS からのデータロードがコピーではなく,単純に移動しているため負荷が少なく,短時間で済むという点です.

  1. 変更したいデータを退避
  2. 変更したい Hive テーブルを削除
  3. 変更した定義で Hive テーブルを作成
  4. ひたすら退避したデータを Hive テーブルに再ロード

変更したいデータを退避

hdfs コマンドを使って,Hive テーブルのデータを移動する.


hdfs dfs -mv /user/hive/warehouse/hoge.db/fuga/ /temporary/hoge.db/

変更したい Hive テーブルを削除

HDFS 的にはデータが移動されていますが,Hive のメタデータが残っています. このメタデータを削除するために,普通に Hive テーブルを削除します.


hive> USE hoge;
hive> DROP TABLE IF EXISTS fuga;

変更した定義で Hive テーブルを作成

新しい Hive テーブルの定義を同じ名前で作成します.

ひたすら退避したデータを Hive テーブルに再ロード

あとは上述した方法と同じように,HDFS 上のパスから Hive テーブルの PARTITION 名を読み取って LOAD DATA します.

このような仕組みを応用することで,ALTER TABLE をかけられなくなってしまった Hive テーブルのスキーマ変更を低コストで実現できます.

おわりに

今回紹介した方法は,かなり無理矢理なデータ移行方法です. CDH3, CDH4 で確認できる Hive のデータロードが,元データに対して「なにかしら」の変換が行われないということから成り立つ方法により「たまたま」できるというだけです. 「どうしても」というときには姑息な手段として,この Hive のデータ移行方法が役に立つかもしれません.