Embulkと日付データ(その2: PostgreSQLへのデータ投入) #Embulk

EmbulkとPostgreSQLと使う場合の注意点メモ

環境

embulk: 0.6.16
embulk-output-postgresql: 0.4.1
PostgreSQL: 9.4

まとめ

PostgreSQLに日付データを入れる場合の注意事項

インサート先のカラムがtimezone型(WITHOUT TIME ZONE)の場合、設定ファイルで忘れずにタイムゾーンを指定する。
未指定の場合、PostgreSQL側のタイムゾーンが日本になっていても、プレビューで表示されたUTCの値がそのまま入ってしまう
timezone型(WITH TIME ZONE)の場合は、embulkがタイムゾーンを考慮してくれる。
Embulkのtimestamp型を、date型に投入する場合はWITHOUT TIME ZONEと同様にtimezoneを指定する。

解説

Embulkと日付データ(その1: CSVの入力データ) で使ったCSVデータをそのまま用いる

CSVパーサーで、default_timezoneを指定した「2015-01-27 19:23:49」のデータは、embulk previewでは、UTC時刻「2015-01-27 10:23:49 UTC」と表示される。

このデータをPostgreSQLに投入した場合の動作を確認する

embulk preview -G config.yaml
*************************** 1 ***************************
      id (     long) : 1
 account (     long) : 32,864
    time (timestamp) : 2015-01-27 10:23:49 UTC
purchase (timestamp) : 2015-01-26 15:00:00 UTC
 comment (   string) : embulk

PostgreSQL側のタイムゾーンの確認

データを投入する先のPostgreSQLのサーバのタイムゾーンはJapanに指定している。

PostgreSQLのタイムゾーン設定は次のSQLで確認できる。
出力された日付が+09と表示されており、日本時間になっていることが確認できる

select now();
              now              
-------------------------------
 2015-07-15 16:17:12.120466+09
(1 row)

PostgreSQLのtimezone型

PostgreSQLのtimestamp型には、「タイムゾーンあり(WITH TIME ZONE)」と「タイムゾーンなしWITHOUT TIME ZONE」という型ががある。
SQL標準の場合、timestamp型はタイムゾーンなしだそうだ。参考

timestamp WITHOUT TIMEZONEにデータを投入する

次のSQLを流して、インサート先のテーブルを作成する。

DROP TABLE IF EXISTS "public"."embulk_example";
CREATE TABLE "public"."embulk_example" (
    "id" int8,
    "account" int8,
    "time" timestamp(6) WITHOUT TIME ZONE,
    "purchase" timestamp(6) WITHOUT TIME ZONE,
    "comment" text COLLATE "default"
)
WITH (OIDS=FALSE);
ALTER TABLE "public"."embulk_example" OWNER TO "postgres";

embulkの設定

次のような設定を用意する。

out:
  type: postgresql
  host: host
  user: user
  password: xxxxx
  database: database
  table: embulk_example
  mode: insert

データの投入

embulk runでデータを投入する

embulk run config.yml 
2015-07-15 16:13:39.142 +0900: Embulk v0.6.16
2015-07-15 16:13:41.364 +0900 [INFO] (transaction): Loaded plugin embulk-output-postgresql (0.4.1)
2015-07-15 16:13:41.414 +0900 [INFO] (transaction): Listing local files at directory '/private/tmp/postgresql_test/pgtest/csv' filtering filename by prefix 'sample_'
2015-07-15 16:13:41.421 +0900 [INFO] (transaction): Loading files [/private/tmp/postgresql_test/pgtest/csv/sample_01.csv.gz]
2015-07-15 16:13:41.520 +0900 [INFO] (transaction): Connecting to jdbc:postgresql://host:5432/database options {user=postgres, tcpKeepAlive=true, loginTimeout=300, socketTimeout=1800}
2015-07-15 16:13:41.609 +0900 [INFO] (transaction): SQL: SET search_path TO "public"
2015-07-15 16:13:41.614 +0900 [INFO] (transaction): > 0.00 seconds
2015-07-15 16:13:41.614 +0900 [INFO] (transaction): Using insert mode
2015-07-15 16:13:41.651 +0900 [INFO] (transaction): SQL: DROP TABLE IF EXISTS "embulk_exam_1f78a400_bl_tmp000"
2015-07-15 16:13:41.656 +0900 [INFO] (transaction): > 0.00 seconds
2015-07-15 16:13:41.670 +0900 [INFO] (transaction): SQL: CREATE TABLE IF NOT EXISTS "embulk_exam_1f78a400_bl_tmp000" ("id" INT8, "account" INT8, "time" TIMESTAMP, "purchase" TIMESTAMP, "comment" TEXT)
2015-07-15 16:13:41.674 +0900 [INFO] (transaction): > 0.00 seconds
2015-07-15 16:13:41.710 +0900 [INFO] (transaction): {done:  0 / 1, running: 0}
2015-07-15 16:13:41.761 +0900 [INFO] (task-0000): Connecting to jdbc:postgresql://host:5432/database options {user=postgres, tcpKeepAlive=true, loginTimeout=300, socketTimeout=1800}
2015-07-15 16:13:41.788 +0900 [INFO] (task-0000): SQL: SET search_path TO "public"
2015-07-15 16:13:41.790 +0900 [INFO] (task-0000): > 0.00 seconds
2015-07-15 16:13:41.792 +0900 [INFO] (task-0000): Copy SQL: COPY "embulk_exam_1f78a400_bl_tmp000" ("id", "account", "time", "purchase", "comment") FROM STDIN
2015-07-15 16:13:41.966 +0900 [INFO] (task-0000): Loading 4 rows (340 bytes)
2015-07-15 16:13:41.990 +0900 [INFO] (task-0000): > 0.02 seconds (loaded 4 rows in total)
2015-07-15 16:13:41.992 +0900 [INFO] (transaction): {done:  1 / 1, running: 0}
2015-07-15 16:13:41.993 +0900 [INFO] (transaction): Connecting to jdbc:postgresql://host:5432/database options {user=postgres, tcpKeepAlive=true, loginTimeout=300, socketTimeout=28800}
2015-07-15 16:13:42.005 +0900 [INFO] (transaction): SQL: SET search_path TO "public"
2015-07-15 16:13:42.008 +0900 [INFO] (transaction): > 0.00 seconds
2015-07-15 16:13:42.009 +0900 [INFO] (transaction): SQL: INSERT INTO "embulk_example" ("id", "account", "time", "purchase", "comment") SELECT "id", "account", "time", "purchase", "comment" FROM "embulk_exam_1f78a400_bl_tmp000"
2015-07-15 16:13:42.013 +0900 [INFO] (transaction): > 0.00 seconds (4 rows)
2015-07-15 16:13:42.030 +0900 [INFO] (transaction): Connecting to jdbc:postgresql://host:5432/database options {user=postgres, tcpKeepAlive=true, loginTimeout=300, socketTimeout=1800}
2015-07-15 16:13:42.045 +0900 [INFO] (transaction): SQL: SET search_path TO "public"
2015-07-15 16:13:42.094 +0900 [INFO] (transaction): > 0.05 seconds
2015-07-15 16:13:42.095 +0900 [INFO] (transaction): SQL: DROP TABLE IF EXISTS "embulk_exam_1f78a400_bl_tmp000"
2015-07-15 16:13:42.099 +0900 [INFO] (transaction): > 0.00 seconds
2015-07-15 16:13:42.100 +0900 [INFO] (main): Committed.
2015-07-15 16:13:42.100 +0900 [INFO] (main): Next config diff: {"in":{"last_path":"/private/tmp/postgresql_test/pgtest/csv/sample_01.csv.gz"},"out":{}}

確認

一番目のエントリは、プレビュー時「2015-01-27 10:23:49 UTC」であり日本時間では、「2015-01-27 19:23:49」であるがインサートするとUTCの時間がそのままインサートされてしまっている。

select * from embulk_example ;
 id | account |        time         |      purchase       |          comment           
----+---------+---------------------+---------------------+----------------------------
  1 |   32864 | 2015-01-27 10:23:49 | 2015-01-26 15:00:00 | embulk
  2 |   14824 | 2015-01-27 10:01:23 | 2015-01-26 15:00:00 | embulk jruby
  3 |   27559 | 2015-01-27 17:20:02 | 2015-01-27 15:00:00 | Embulk "csv" parser plugin
  4 |   11270 | 2015-01-29 02:54:36 | 2015-01-28 15:00:00 | NULL

タイムゾーンの指定

このような場合out側のプラグインの設定でタイムゾーンを指定する。
タイムゾーンは、「カラム毎の設定」または「デフォルトのタイムゾーン指定」をすることができる。

カラム毎の設定の場合はcolumn_optionsでタイムゾーンを指定する。この場合時刻を利用するカラム毎に設定が必要。

out:
  type: postgresql
  host: host
  user: user
  password: xxxxx
  database: database
  table: embulk_example
  mode: insert
  column_options:                   # <-- これ
    time: { timezone: "Japan" }     # <-- timeカラム用
    purchase: { timezone: "Japan" } # <-- purchaseカラム用

すべてのカラムで指定する必要がない場合は、default_timezoneを指定すると良い

out:
  type: postgresql
  host: host
  user: user
  password: xxxxx
  database: database
  table: embulk_example
  mode: insert
  default_timezone: "Japan"  # <-- これ

これを指定して再度データをインサートすると、日本時間に変換された時刻が各カラムにインサートされることが確認できる

select * from embulk_example ;
 id | account |        time         |      purchase       |          comment           
----+---------+---------------------+---------------------+----------------------------
  1 |   32864 | 2015-01-27 19:23:49 | 2015-01-27 00:00:00 | embulk
  2 |   14824 | 2015-01-27 19:01:23 | 2015-01-27 00:00:00 | embulk jruby
  3 |   27559 | 2015-01-28 02:20:02 | 2015-01-28 00:00:00 | Embulk "csv" parser plugin
  4 |   11270 | 2015-01-29 11:54:36 | 2015-01-29 00:00:00 | NULL
(4 rows)

timestamp WITH TIME ZONEにデータを投入する

DROP TABLE IF EXISTS "public"."embulk_example";
CREATE TABLE "public"."embulk_example" (
    "id" int8,
    "account" int8,
    "time" timestamp(6) WITH TIME ZONE,
    "purchase" timestamp(6) WITH TIME ZONE,
    "comment" text COLLATE "default"
)
WITH (OIDS=FALSE);
ALTER TABLE "public"."embulk_example" OWNER TO "postgres";

PostgreSQL固有のtimestamp WITH TIMEZONE型の場合、default_timezoneやcolumn_optionsを指定しなくても、タイムゾーンを考慮してデータをインサートしてくれる。

mode: insert_directでembulkがテーブルを自動作成する場合、デフォルトではタイムゾーンありの型で作成される。

 id | account |          time          |        purchase        |          comment           
----+---------+------------------------+------------------------+----------------------------
  1 |   32864 | 2015-01-27 19:23:49+09 | 2015-01-27 00:00:00+09 | embulk
  2 |   14824 | 2015-01-27 19:01:23+09 | 2015-01-27 00:00:00+09 | embulk jruby
  3 |   27559 | 2015-01-28 02:20:02+09 | 2015-01-28 00:00:00+09 | Embulk "csv" parser plugin
  4 |   11270 | 2015-01-29 11:54:36+09 | 2015-01-29 00:00:00+09 | NULL
(4 rows)