embulk unpivot filterプラグイン作りました。

データ転送ツールのEmbulkは大量データの投入に威力を発揮しますが、filter pluginを組み合わせることによってかなり複雑な処理も実現できるため、yamlで書けるETLツールとしても優秀です。

今回RDBMSで正規化されたテーブルにデータを投入する必要ができたので、unpivot処理を行うpluginを書きましたので公開しておきます。

github: embulk-filter-unpivot

インストール

embulk gem install embulk-filter-unpivot

でインストールできます。

どんなプラグインか

一言で言うと「横持ち」のテーブル構造から「縦持ち」のテーブルに変換するフィルタです。

例えば、不動産物件のCSVデータがこのようにあるとします。

• (A)変換前のデータ(横持ちデータ)

物件ID	家賃	物件情報入力日	ペット可	駐車場付き	バス・トイレ別	エアコン
1	8.5	2017-05-01	0	0	1	1
2	5.5	2017-05-01	0	0	0	0
3	10.0	2017-05-02	1	0	1	1
4	15.5	2017-05-02	0	1	1	1

物件の属性(バス・トイレ別やエアコンの有無)が0/1で表現されています。ただ物件の属性というのは数百種類以上になることもあり、かつ「駐車場付き」などは殆どが0のカラムとなるので、データ構造としては無駄の多いテーブルとなります(こういうデータをスパースデータなどと呼びます)。
そこでRDBMSではこの部分を別テーブルに移して以下のように表現することがよく行われます。

• (B)縦持ちデータ

物件ID	設備名
1	バス・トイレ別
1	エアコン
3	ペット可
3	バス・トイレ別
3	エアコン
4	駐車場付き
4	バス・トイレ別
4	エアコン

物件IDと存在する(1が立っている)カラムだけのレコードを作ることで、レコード数は多いもののカラム数は2つだけのテーブルにできます。存在しない(値が0の)カラムのレコードは作りませんので、スパースデータの場合、かなり全体のサイズが圧縮可能です。
さらに、設備名にIDをつけて、

• (C)設備コード表

設備ID	設備
1	ペット可
2	駐車場付き
3	バス・トイレ別
4	エアコン

文字列をこのコードに置き換えると以下のようになります。

• (D)縦持ちデータ(コード化後)

物件ID	設備ID
1	3
1	4
3	1
3	3
3	4
4	2
4	3
4	4

データサイズはかなり小さくなりました。

横持ちデータを縦持ちデータに変換する処理をunpivot、その逆をpivot処理と呼びます。Excelのピボットテーブルを思い出していただくと良いかと思います。

今回のプラグインはこのunpivotを行います。

書き方

上の例のデータ(A)から(D)に変換するfilterの例です。

filters:
  - type: column
    add_columns:
      - {name: 設備ID, type: long, default: 0}
  - type: unpivot
    outer_key: 物件ID
    inner_key: 設備ID
    columns:
      - {name: 'ペット可', id: 1}
      - {name: '駐車場付き', id: 2}
      - {name: 'バス・トイレ別', id: 3}
      - {name: 'エアコン', id: 4}

unpivotプラグイン自体はカラムを定義したり変換したりする機能を持たないので、まず必要なカラムをcolumnプラグインで定義しておきます。最終的に必要なカラムは「物件ID」と「設備ID」ですが、データ(A)に設備IDはないので、add_columnsで追加します。

その後、type: unpivotでプラグインの利用を宣言し、outer_keyにレコードのID(ここでは物件ID）を、inner_keyに「設備ID」を指定します。

さらに、columnsで(C)設備コード表に当たるkey/valueを定義します。idは今回int型ですが、stringで定義することも可能です。コード表を持たせず(B)のようなデータを作りたい場合は- {name: 'ペット可', id:ペット可`}と書けばOKです。

yaml内にin: とout: を定義してembulkを実行すると(D)のデータが出来上がります。

カラムを追加する場合

全レコードに同じカラムを追加したい場合は「additional」というキーが利用できます。

filters:
  - type: column
    add_columns:
      - {name: 設備ID, type: long, default: 0}
      - {name: created_at, type: timestamp, src: '物件情報入力日'}
      - {name: updated_at, type: timestamp, src: '物件情報更新日'}
  - type: unpivot
    outer_key: 物件ID
    inner_key: 設備ID
    additional: [created_at, updated_at]
    columns:
      - {name: 'ペット可', id: 1}
      - {name: '駐車場付き', id: 2}
      - {name: 'バス・トイレ別', id: 3}
      - {name: 'エアコン', id

additionalには各レコードの後ろに付与するカラム名を定義します。配列で複数もつことも可能です。上記の例だと、created_atとupdate_atというカラムが、物件情報の入力日と更新日が入った状態で全レコードに追加されます。なお、outer_key/inner_key/additionalで指定したカラム以外は全て削除されます。

なお、unpibvot後のレコードを生成するかどうか(フラグが立っているとみなすかどうか)は、「0または空文字('')以外」という条件式にしていますので、変換前のデータが厳密に「0/1」でなくとも構いません。