embulk使ってみました。そして、速さを求めてpg_bulkloadとの組み合わせにトライしてみました。

使ったバージョンは

embulk : 0.7.5
PostgreSQL : 9.5alpha2

embulkのインストール

公式ドキュメント通りにコマンドを叩けばOKです。

PostgreSQL出力用プラグイン

以下のコマンドでインストールします。

$ embulk gem install embulk-output-postgresql

PostgreSQL JDBC Driverはバンドルしているようなので、追加配置は不要です。
build.gradleを見るとバージョンは9.4-1200-jdbc41。

ロード

今回は、CSVファイルのデータをembulkを使ってロードするのを試してみます。

対象テーブル

ちょーシンプルなテーブルで試します。

$ psql
psql (9.5alpha2)
Type "help" for help.

[local] 33914 postgres=# \d tbl
      Table "public.tbl"
 Column |  Type   | Modifiers
--------+---------+-----------
 a      | integer | not null
 b      | text    |
Indexes:
    "tbl_pkey" PRIMARY KEY, btree (a)

ロードするデータはこんな感じ。

$ head  /home/postgres/tmp/dump.tbl
687805,000002b3ee394b36e8323d7d3e5f29fb
751689,0000114d140790801fb40608048c76e6
297583,0000246baf5fa28b1b3f369f3ac5eade
710805,00002e712863c0ad4ad631220909609b
250149,0000324faa31156eadf83d8a85a9bbd6
956774,00003389fdbd39415cc63f044d01599c
900235,00004a1b08ae3655504460663f52407d
66635,00006b6e7e964c78cad2658aefd97683
50086,000070af5c6860e0baad557cfac1f67a
876796,00007757c74ead14d87a113a576f5eed

1,000,000行あります。
データはPostgreSQLでgenerated_seriesを使ってランダムな値が並ぶように用意しました。

embulkの設定ファイル

マニュアルをみつつ作成。

jdbc_config.yml

$ cat jdbc_config.yml
in:
  type: file
  path_prefix: /home/postgres/tmp/dump
  parser:
    charset: UTF-8
    newline: CRLF
    type: csv
    delimiter: ','
    quote: '"'
    trim_if_not_quoted: false
    allow_extra_columns: false
    allow_optional_columns: false
    columns:
    - {name: a, type: long}
    - {name: b, type: string}
out:
   type: postgresql
   host: localhost
   user: postgres
   database: postgres
   table: tbl
   mode: insert

ちなみに、インプットのカラムa用データの型をintと指定したら、そんなのないよとembulkに怒られたのでlongにしています。

Error: com.fasterxml.jackson.databind.JsonMappingException: Unknown type name 'int'. Supported types are: boolean, long, double, string, timestamp (through reference chain: java.util.ArrayList[0])

ロード実行

$ time embulk run jdbc_config.yml
2015-10-10 23:41:15.431 +0900: Embulk v0.7.5
2015-10-10 23:41:18.028 +0900 [INFO] (transaction): Loaded plugin embulk-output-postgresql (0.4.1)
++++++++++++ 略 +++++++++++++++++++++++++++++
2015-10-10 23:41:18.495 +0900 [INFO] (transaction): SQL: CREATE TABLE IF NOT EXISTS "tbl_5619238d34b0c500_bl_tmp000" ("a" INT4, "b" TEXT)
++++++++++++ 略 +++++++++++++++++++++++++++++
2015-10-10 23:41:18.662 +0900 [INFO] (task-0000): Copy SQL: COPY "tbl_5619238d34b0c500_bl_tmp000" ("a", "b") FROM STDIN
++++++++++++ 略 +++++++++++++++++++++++++++++
2015-10-10 23:41:26.818 +0900 [INFO] (transaction): SQL: INSERT INTO "tbl" ("a", "b") SELECT "a", "b" FROM "tbl_5619238d34b0c500_bl_tmp000"
++++++++++++ 略 +++++++++++++++++++++++++++++
2015-10-10 23:41:33.528 +0900 [INFO] (transaction): SQL: DROP TABLE IF EXISTS "tbl_5619238d34b0c500_bl_tmp000"
++++++++++++ 略 +++++++++++++++++++++++++++++
2015-10-10 23:41:33.554 +0900 [INFO] (main): Committed.
2015-10-10 23:41:33.555 +0900 [INFO] (main): Next config diff: {"in":{"last_path":"/home/postgres/tmp/dump.tbl"},"out":{}}

real    0m23.407s
user    0m23.660s
sys     0m4.673s

insertモードで実行したので、一時テーブルにデータがCOPY文でデータがロードされ、その後目的のテーブルにSELECT INSERTされます。

insert_directモードにすると、一時テーブルを使わず、直接COPY文で対象テーブルにデータがロードされます。

$ time embulk run jdbc_config.yml
2015-10-10 23:47:26.830 +0900: Embulk v0.7.5
2015-10-10 23:47:29.359 +0900 [INFO] (transaction): Loaded plugin embulk-output-postgresql (0.4.1)
++++++++++++ 略 +++++++++++++++++++++++++++++
2015-10-10 23:47:29.866 +0900 [INFO] (task-0000): Copy SQL: COPY "tbl" ("a", "b") FROM STDIN
++++++++++++ 略 +++++++++++++++++++++++++++++
2015-10-10 23:47:42.501 +0900 [INFO] (main): Committed.
2015-10-10 23:47:42.502 +0900 [INFO] (main): Next config diff: {"in":{"last_path":"/home/postgres/tmp/dump.tbl"},"out":{}}

real    0m20.968s
user    0m24.651s
sys     0m3.912s

少し早くなりますが、insert_directモードだとロード中にエラーが起きてembulkが停止した場合に、対象テーブルに中途半端なデータがロードされたまま残ったりするなどのリスクがあるようです。

さらに高速にロードしたい！

PostgreSQLに高速にデータをロードするツールとして、pg_bulkloadがあります。

こいつをembulkと組み合わせてみましょう！

pg_bulkload

pg_bulkloadは

SQL経由ではなく、PostgreSQLの内部関数を直接呼び出してレコードをテーブルファイルに直接書き込み
トランザクションログの出力をなるべくスキップして、書き込み処理の負荷を削減
インデックス更新処理を追加レコードごとに実施せず、最後にまとめて実施
パースと書き込みを別プロセスで並列実行（オプション指定時）

という反則技でロード高速化を実現しています。

もちろん、上記のような反則を使うのでpg_bulkloadで行ったデータロードの操作はトランザクションログからは再現できないという制約があります。また、ロード中にpg_bulkloadがクラッシュした場合に特別なリカバリが必要になったりします。

このような制約を守るのであれば、他のツールでは得ることができないロード性能を得ることができるので、用法を守って正しく使うのであれば大変有用です。

なんというか、ラディカルグッドスピード（古い）

pg_bulkloadのインストール

今回は手元の環境でPostgreSQL 9.5alpha2を使っていたので、GitHubからソースコードを入手してインストールします。9.4以前ならば、RPMがあるのでそちらを使ってインストールしても大丈夫です。

ソースコードからビルドする場合、必要となるライブラリがいくつかあります。
基本的にはPostgreSQLをソースコードからビルドして動かしている環境であれば大丈夫です。
PostgreSQLをソースコードからビルドする方法は過去の記事に書きました

VERSION3_1_STABLEブランチを使います。

$ git clone https://github.com/ossc-db/pg_bulkload.git
$ cd pg_bulkload
$ git checkout -b VERSION3_1_STABLE origin/VERSION3_1_STABLE
$ make
# make install

後は、ロードする対象のデータベースにpg_bulkloadを登録してあげます。

$ psql
psql (9.5alpha2)
Type "help" for help.

[local] 26537 postgres=# CREATE EXTENSION pg_bulkload ;

これでOKです。

embulk-output-command

調べたところ、embulkの出力プラグインとして任意のコマンドを実行することができるembulk-output-commandがあることを発見。

一方、pg_bulkloadは標準入力からデータを受け取ることができるので、この2つを組み合わせることができそうです。

$ embulk gem install embulk-output-command

でインストール。

embulk便利だわ。

embulk + pg_bulkloadのためのconfig.yml

以下のようにすれば良いことが分かりました。

bulk_config.yml

$ cat bulk_config.yml
in:
  type: file
  path_prefix: /home/postgres/tmp/dump
  parser:
    charset: UTF-8
    newline: CRLF
    type: csv
    delimiter: ','
    quote: '"'
    trim_if_not_quoted: false
    allow_extra_columns: false
    allow_optional_columns: false
    columns:
    - {name: a, type: long}
    - {name: b, type: string}
out:
   type: command
   command: "pg_bulkload -i stdin -O tbl -o SKIP=1"
   formatter:
     type: csv

inのところは、embulk-output-postgresqlの時と同じです。
outのところに、embulk-output-commandプラグインを指定して、実行するコマンドとしてpg_bulkloadを呼んであげます。

pg_bulkloadのコマンドオプションとして、-iで標準入力からデータを受けつけるようにします。-Oでロード先テーブルを指定します。
また、-oでSKIP=1として、最初の1行を読み飛ばすようにします。これは、embulk-output-commandを使った場合、最初の１行としてカラム名が並んだレコードが出力されるためです。

ロード！

$ time embulk run bulk_config.yml
2015-10-11 00:29:18.492 +0900: Embulk v0.7.5
2015-10-11 00:29:20.861 +0900 [INFO] (transaction): Loaded plugin embulk-output-command (0.1.3)
2015-10-11 00:29:20.917 +0900 [INFO] (transaction): Listing local files at directory '/home/postgres/tmp' filtering filename by prefix 'dump'
2015-10-11 00:29:20.925 +0900 [INFO] (transaction): Loading files [/home/postgres/tmp/dump.tbl]
2015-10-11 00:29:21.071 +0900 [INFO] (transaction): {done:  0 / 1, running: 0}
2015-10-11 00:29:21.097 +0900 [INFO] (task-0000): Using command [sh, -c, pg_bulkload -i stdin -O tbl -o SKIP=1]
NOTICE: BULK LOAD START
NOTICE: BULK LOAD END
        1 Rows skipped.
        1000000 Rows successfully loaded.
        0 Rows not loaded due to parse errors.
        0 Rows not loaded due to duplicate errors.
        0 Rows replaced with new rows.
2015-10-11 00:29:25.787 +0900 [INFO] (transaction): {done:  1 / 1, running: 0}
2015-10-11 00:29:25.795 +0900 [INFO] (main): Committed.
2015-10-11 00:29:25.796 +0900 [INFO] (main): Next config diff: {"in":{"last_path":"/home/postgres/tmp/dump.tbl"},"out":{}}

real    0m12.006s
user    0m21.005s
sys     0m2.335s

手元の環境では、embulk-output-postgresqlの時間(平均24秒くらい)よりも半分くらいの時間(平均12秒くらい)でロードが出来るようになりました(≧∇≦)/

まとめ

少し触っただけですが、embulk使いやすくていいです！

いろんなプラグインがあるので、各種データ・ソースの間でデータを移動させるのに便利そうです。

embulkでPostgreSQLにデータをロードする際に「もっと速さを！！」という状態になったら、pg_bulkloadを組み合わせるこの技を使ってみてください。

参考

pg_bulkload ドキュメント
http://ossc-db.github.io/pg_bulkload/pg_bulkload-ja.html

Embulk + pg_bulkloadでPostgreSQLに高速データロード！