Apache Kudu (http://kudu.apache.org/) がApacheプロジェクトのTLP（トップレベルプロジェクト）になったので、最新のQuickstart VMを使って試してみました。

Apache Kuduとは

Apache Kuduは一から開発された全く新しいオープンソースのストレージエンジンです。Hadoopエコシステムに位置付けられ、HDFSとHBaseを補完します。Kuduにより、頻繁に追加/更新されるデータを素早くクエリすることができるようになります。例えば一般的なデータベースのように、リアルタイムでインサート、更新、削除などのデータ操作が可能です。

Kuduのブログより抜粋 http://kudu.apache.org/2016/07/25/asf-graduation.html

Optimized for lightning-fast scans, Kudu is particularly well suited to hosting time-series data and various types of operational data. In addition to its impressive scan speed, Kudu supports many operations available in traditional databases, including real-time insert, update, and delete operations. Kudu enables a “bring your own SQL” philosophy, and supports being accessed by multiple different query engines including such other Apache projects as Drill, Spark, and Impala (incubating).

ドキュメント (http://kudu.apache.org/overview.html#architecture) によると、
- Super Fast Columnar Storage (超高速列指向ストレージ）で
- 分散型で耐障害性を持ち
- Intel 3D Xpointのような次世代ストレージに対応
しています

KuduはHadoopのHDFSやHBaseを置き換えるものではなく、これらを補完するものです。例えば大量データのスキャンであればHDFS、ランダムread/writeであればHBaseの方が高速です。しかしHDFSやHBaseは、頻繁に更新されるデータを素早く分析する用途にはあまり適していません。
KuduはApache Parquetのように列指向でデータを扱います。HBaseとは異なり、各列はそれぞれ型を持つのでデータを効率良くエンコード、圧縮できます。分析時に一部の列にアクセスするような場合、全てのデータを読み書きする必要はなく、ディスクI/Oが少なくて済むため高速です。Kuduはデータの追加、更新、削除ができるだけではなく、列へのアクセスが効率的なカラムナストレージエンジンということです。

参考情報
- Kudu (クドゥ) : HDFSとHBaseを補完する、Hadoopの新しいストレージエンジン
- Kudu情報まとめ

Quickstart VM

Kuduの公式ページからVirtual Box上で動作する Quickstart VMが利用できます。さすがに分散環境ではありませんが、Kuduを手っ取り早く試してみるにはいいですね。
http://kudu.apache.org/docs/quickstart.html

実行環境

Macbook Pro (OSX El Capitan)
VirtualBox 5.0.10

ブートストラップの実行

Kuduのブートストラップを取得して実行します。Mac上でターミナルを開いて下記のコマンドを実行します。

$ curl -s https://raw.githubusercontent.com/cloudera/kudu-examples/master/demo-vm-setup/bootstrap.sh | bash

自動的にVirtual Boxのイメージのダウンロードが始まります。1.2GB程度あるのでしばらく時間がかかります。

$ curl -s https://raw.githubusercontent.com/cloudera/kudu-examples/master/demo-vm-setup/bootstrap.sh | bash
+ : https://github.com/cloudera/kudu-examples.git
+ git clone https://github.com/cloudera/kudu-examples.git
Cloning into 'kudu-examples'...
remote: Counting objects: 366, done.
remote: Total 366 (delta 0), reused 0 (delta 0), pack-reused 366
Receiving objects: 100% (366/366), 69.88 KiB | 0 bytes/s, done.
Resolving deltas: 100% (119/119), done.
Checking connectivity... done.
+ pushd kudu-examples
~/Downloads/kudu-examples ~/Downloads
+ pushd demo-vm-setup
~/Downloads/kudu-examples/demo-vm-setup ~/Downloads/kudu-examples ~/Downloads
+ ./setup-kudu-demo-vm.sh
Downloading Virtualbox Image file: http://cloudera-kudu-beta.s3.amazonaws.com/cloudera-quickstart-vm-5.7.1-kudu-virtualbox.ova
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100 1361M  100 1361M    0     0  3069k      0  0:07:34  0:07:34 --:--:-- 1138k

ダウンロードが終わるとVirtualBox上で仮想マシンが起動します。

100 1361M  100 1361M    0     0  3216k      0  0:07:13  0:07:13 --:--:-- 3892k
0%...10%...20%...30%...40%...50%...60%...70%...80%...90%...100%
Interface 'vboxnet6' was successfully created
Importing VM cloudera-quickstart-vm-5.7.1-kudu-virtualbox.ova...
0%...10%...20%...30%...40%...50%...60%...70%...80%...90%...100%
Interpreting /Users/kawasaki/Downloads/kudu-examples/demo-vm-setup/cloudera-quickstart-vm-5.7.1-kudu-virtualbox.ova...
OK.
Disks:
  vmdisk1   68719476736 -1  http://www.vmware.com/interfaces/specifications/vmdk.html#streamOptimized   cloudera-quickstart-vm-5.7.1-kudu-virtualbox-disk1.vmdk -1  -1  

Virtual system 0:
 0: Suggested OS type: "RedHat_64"
    (change with "--vsys 0 --ostype <type>"; use "list ostypes" to list all possible values)
 1: VM name specified with --vmname: "kudu-demo"
 2: No. of CPUs specified with --cpus: 2
 3: Guest memory specified with --memory: 6144 MB
 4: Network adapter: orig NAT, config 3, extra slot=0;type=NAT
 5: Network adapter: orig NAT, config 3, extra slot=1;type=NAT
 6: CD-ROM
    (disable with "--vsys 0 --unit 6 --ignore")
 7: IDE controller, type PIIX4
    (disable with "--vsys 0 --unit 7 --ignore")
 8: IDE controller, type PIIX4
    (disable with "--vsys 0 --unit 8 --ignore")
 9: Hard disk image: source image=cloudera-quickstart-vm-5.7.1-kudu-virtualbox-disk1.vmdk, target path=/Users/kawasaki/VirtualBox VMs/cloudera-quickstart-vm-5.7.1-kudu-virtualbox/cloudera-quickstart-vm-5.7.1-kudu-virtualbox-disk1.vmdk, controller=7;channel=0
    (change target path with "--vsys 0 --unit 9 --disk path";
    disable with "--vsys 0 --unit 9 --ignore")
0%...10%...20%...30%...40%...50%...60%...70%...80%...90%...100%
Successfully imported the appliance.
Waiting for VM "kudu-demo" to power on...
VM "kudu-demo" has been successfully started.
Wait until services become available.
Updating the /etc/hosts file requires sudo rights.
Password:

スクリーンショット

終わったら別のターミナルを開いてsshでアクセスしてみましょう。手元の環境では以前に試した環境が残っていたせいか、quickstart.cloudera に通信できなかったので /etc/hosts に下記のように追加しました。

127.0.0.1   localhost   localhost.domain
192.168.62.100  quickstart.cloudera quickstart

sshのユーザー名とパスワードは demo です。

$ ssh demo@quickstart.cloudera
demo@quickstart.cloudera's password: 
Last login: Tue Jul 26 05:16:28 2016 from 192.168.60.1
[demo@quickstart ~]$

無事にログインできました。kuduとimpaladが動作していることを確認します。

[demo@quickstart ~]$ ps aux |grep kudu
[demo@quickstart ~]$ ps aux |grep kudu
kudu      3035  0.7  0.5 547028 31164 ?        Sl   05:51   0:01 /usr/lib/kudu/sbin/kudu-master --flagfile=/etc/kudu/conf/master.gflagfile
kudu      3174  0.6  0.2 1245340 13968 ?       Sl   05:51   0:01 /usr/lib/kudu/sbin/kudu-tserver --flagfile=/etc/kudu/conf/tserver.gflagfile
demo      3811  0.0  0.0 107452   948 pts/0    S+   05:55   0:00 grep kudu
[demo@quickstart ~]$ ps aux |grep impalad
impala    2999  2.4  2.7 4170804 163588 ?      Sl   05:51   0:05 /usr/lib/impala/sbin/impalad -log_dir=/var/log/impala -catalog_service_host=127.0.0.1 -state_store_port=24000 -use_statestore -state_store_host=127.0.0.1 -be_port=22000
demo      3813  0.0  0.0 107452   948 pts/0    S+   05:55   0:00 grep impalad
[demo@quickstart ~]$

テスト

ダウンロードページのチュートリアルに従って進めます。

サンプルデータのロード

SFO Passenger Dataをロードします。

$ hdfs dfs -mkdir /data
$ hdfs dfs -put examples/SFO_Passenger_Data/MonthlyPassengerData_200507_to_201506.csv /data

impala shellの起動

分散SQLエンジンのImpalaを使ってKuduに接続します。別のターミナルを開いて impala-shell を開始します。。

$ ssh demo@quickstart.cloudera -t impala-shell
demo@quickstart.cloudera's password: 
Starting Impala Shell without Kerberos authentication
Connected to quickstart.cloudera:21000
Server version: impalad version 2.7.0-cdh5-IMPALA_KUDU-cdh5 RELEASE (build fc36c3c7fbbbdfb0e8b1b0e6ee7505531a384550)
***********************************************************************************
Welcome to the Impala shell. Copyright (c) 2015 Cloudera, Inc. All rights reserved.
(Impala Shell v2.7.0-cdh5-IMPALA_KUDU-cdh5 (fc36c3c) built on Tue Jun 14 23:55:45 PDT 2016)

To see live updates on a query's progress, run 'set LIVE_SUMMARY=1;'.
***********************************************************************************
[quickstart.cloudera:21000] >

テーブルの作成

impala shell から以下をコピペして passenger_data_raw テーブルを作成します。これはHDFS上にアップロードしたデータを使う、一般的なImpala/Hiveテーブルです。

CREATE EXTERNAL TABLE passenger_data_raw (
  id int,
  activity_period int,
  operating_airline string,
  airline_iata_code string,
  published_airline string,
  published_airline_iata_code string,
  geo_summary string,
  geo_region string,
  activity_type_code string,
  price_category_code string,
  terminal string,
  boarding_area string,
  passenger_count bigint
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
LOCATION '/data/';

テストクエリの実行

クエリしてみます。（一回目はメタデータのキャッシュがされていないので、いくぶん時間がかかります。二回目は0.5s程度でした）

> SELECT count(*) FROM passenger_data_raw;
Query: select count(*) FROM passenger_data_raw
+----------+
| count(*) |
+----------+
| 13901    |
+----------+
Fetched 1 row(s) in 5.69s

Kudu にテーブルを作成

CREATE TABLE AS SELECT を使用して、HDFS上の外部テーブルのデータを用いてKuduにテーブルを作成してみます。が、エラーが返ってきてしまいました。

> CREATE TABLE passenger_data
TBLPROPERTIES(
'storage_handler' = 'com.cloudera.kudu.hive.KuduStorageHandler',
'kudu.table_name' = 'passenger_data',
'kudu.master_addresses' = '127.0.0.1',
'kudu.key_columns' = 'id'
 ) AS SELECT * FROM passenger_data_raw;
Query: create TABLE passenger_data
TBLPROPERTIES(
'storage_handler' = 'com.cloudera.kudu.hive.KuduStorageHandler',
'kudu.table_name' = 'passenger_data',
'kudu.master_addresses' = '127.0.0.1',
'kudu.key_columns' = 'id'
 ) AS SELECT * FROM passenger_data_raw
ERROR: AnalysisException: A data distribution must be specified using a DISTRIBUTE BY clause.

エラーはとDISTRIBUTE BY を指定する必要があると書かれています。ドキュメントによると、DISTRIBUTE BY RANGE または DISTRIBUTE BY HASH のいずれかが必要です。
http://www.cloudera.com/documentation/betas/kudu/0-9-0/topics/kudu_impala.html

今回はRANGE指定ではなくHASHを指定することにし、元のテーブル(passenger_data_raw)のid列を使用して16個のバケットに分けることにしました。

DISTRIBUTE BY HASH (id) INTO 16 BUCKETS

警告が出ていますが、以下のように作成されました。

> CREATE TABLE passenger_data
                            > DISTRIBUTE BY HASH (id) INTO 16 BUCKETS
                            > TBLPROPERTIES(
                            > 'storage_handler' = 'com.cloudera.kudu.hive.KuduStorageHandler',
                            > 'kudu.table_name' = 'passenger_data',
                            > 'kudu.master_addresses' = '127.0.0.1',
                            > 'kudu.key_columns' = 'id'
                            >  ) AS SELECT * FROM passenger_data_raw;
Query: create TABLE passenger_data
DISTRIBUTE BY HASH (id) INTO 16 BUCKETS
TBLPROPERTIES(
'storage_handler' = 'com.cloudera.kudu.hive.KuduStorageHandler',
'kudu.table_name' = 'passenger_data',
'kudu.master_addresses' = '127.0.0.1',
'kudu.key_columns' = 'id'
 ) AS SELECT * FROM passenger_data_raw
+-----------------------+
| summary               |
+-----------------------+
| Inserted 13901 row(s) |
+-----------------------+
WARNINGS: Error converting column: 12 TO BIGINT (Data is: Other)
file: hdfs://quickstart.cloudera:8020/data/MonthlyPassengerData_200507_to_201506.csv
record: 
11381,201309,"Atlas Air, Inc",5Y,"Atlas Air, Inc",5Y,Domestic,US,Deplaned,Other,Other,Other,65
Error converting column: 12 TO BIGINT (Data is: Other)
file: hdfs://quickstart.cloudera:8020/data/MonthlyPassengerData_200507_to_201506.csv
record: 
11382,201309,"Atlas Air, Inc",5Y,"Atlas Air, Inc",5Y,Domestic,US,Thru / Transit,Other,Other,Other,3

Fetched 1 row(s) in 1.86s
[quickstart.cloudera:21000] >

Kudu上のデータをクエリする

Kudu上に作成したテーブルをクエリします。先ほど実行したクエリ同様、一回目はHiveメタストアのデータがキャッシュされていないので時間がかかりますが、二回目以降は早いです。

> SELECT sum(passenger_count) AS total, operating_airline FROM passenger_data
  GROUP BY operating_airline
  HAVING total IS NOT null
  ORDER BY total DESC LIMIT 10;
Query: select sum(passenger_count) AS total, operating_airline FROM passenger_data
  GROUP BY operating_airline
  HAVING total IS NOT null
  ORDER BY total DESC LIMIT 10
+-----------+----------------------------------+
| total     | operating_airline                |
+-----------+----------------------------------+
| 105363917 | United Airlines - Pre 07/01/2013 |
| 51319845  | United Airlines                  |
| 32657456  | SkyWest Airlines                 |
| 31727343  | American Airlines                |
| 23801507  | Delta Air Lines                  |
| 23685267  | Virgin America                   |
| 22507320  | Southwest Airlines               |
| 16235520  | US Airways                       |
| 11860630  | Alaska Airlines                  |
| 6706438   | JetBlue Airways                  |
+-----------+----------------------------------+
Fetched 10 row(s) in 0.46s
[quickstart.cloudera:21000] >

データを更新する

ここからがKuduの本領発揮です。先ほどのデータセットには問題があり同じ航空会社が違う名前で表示されてしまっています。UPDATE を使用してデータを更新します。

> UPDATE passenger_data
                            >   SET operating_airline="United Airlines"
                            >   WHERE operating_airline LIKE "United Airlines - Pre%";
Query: update passenger_data
  SET operating_airline="United Airlines"
  WHERE operating_airline LIKE "United Airlines - Pre%"

Fetched 0 row(s) in 0.15s
[quickstart.cloudera:21000] >

何事もなく更新されたように見えます。クエリを実行して確認してみましょう。

> SELECT sum(passenger_count) AS total, operating_airline FROM passenger_data
                            >   GROUP BY operating_airline
                            >   HAVING total IS NOT null
                            >   ORDER BY total DESC LIMIT 10;
Query: select sum(passenger_count) AS total, operating_airline FROM passenger_data
  GROUP BY operating_airline
  HAVING total IS NOT null
  ORDER BY total DESC LIMIT 10
+-----------+--------------------+
| total     | operating_airline  |
+-----------+--------------------+
| 156683762 | United Airlines    |
| 32657456  | SkyWest Airlines   |
| 31727343  | American Airlines  |
| 23801507  | Delta Air Lines    |
| 23685267  | Virgin America     |
| 22507320  | Southwest Airlines |
| 16235520  | US Airways         |
| 11860630  | Alaska Airlines    |
| 6706438   | JetBlue Airways    |
| 6266220   | Northwest Airlines |
+-----------+--------------------+
Fetched 10 row(s) in 0.46s
[quickstart.cloudera:21000] >

正しくデータが更新されていますね！ HDFSはWrite Onceなファイルシステムなのでデータの更新には課題がありますが、Kuduだとこんなに簡単にデータの更新ができ、頻繁に更新されるデータに対して分析を素早く行うことができます。