Embulkで本番DBをマスキングコピー【Docker開発環境】

DockerでEmbulk環境を作り、本番DBのデータを個人情報をマスクして開発DBにコピーする方法です。
数テーブルであれば簡単にできたので参考になれば。

# 背景
開発DBが開発途中の不整合なデータでしっちゃかめっちゃかになっているという課題があり、本番っぽいデータを作るため本番DBから個人情報をマスクしたデータをコピーすることにしました。
背景として、チームで共通のRDS開発DBを使っています。

コピーするテーブルは個人情報を含むユーザーテーブルその他の数テーブルです。多くても１ヶ月に１度の実行頻度で良かったため、今回は定期実行は考えません。

本番実行はAWS ECSで行ったため、第２弾としてそちらも書ければと思っています。

# 使用技術
Docker
Embulk v0.9.18

# 概要
本番DBも開発DBもRDS for Mysqlの想定です。
AWSのETLサービスとして、GlueやData Pipelineも存在しますが、数テーブルの１度きりの実行で良かったことと、マスキング処理をある程度自由度を持って行いたかったため、今回はDocker（ECS)でEmbulkを使うことにしました。

マスキング処理については、公開されているEmbulkのFilterプラグインに「*」でマスクできるものもあったのですが、今回はそれっぽい値を入れたかったため独自で実装を行っています。

# ディレクトリ構造
完成後のディレクトリ構造

```title:ディレクトリ構造
.
├── Dockerfile
├── entrypoint.sh
└── src
    ├── bundle_dir
    │   ├── Gemfile
    │   ├── Gemfile.lock
    │   ├── embulk
    │   └── jruby
    ├── user_config.yml.liquid
    ├── table2_config.yml.liquid
    ├── table3_config.yml.liquid
    └── config_commons
        ├── _in_config.yml.liquid
        └── _out_config.yml.liquid
```

src直下にコピーするテーブルの数だけEmbulkのconfigファイルを用意します。Dockerfileの最後にentrypoint.shを実行してそれらの定義を元にコピーを実行します。
`./src/bundle_dir`は自作Filterプラグインの実装関連の場所です。（詳しくは後述します）

# bundle_dir
```terminal:terminal
$ mkdir src
$ cd src
$ embulk mkbundle bundle_dir
```

`$ embulk mkbundle bundle_dir` でbundle_dir以下が作られます

参考： [embulk mkbundle または embulk bundle の使い方](https://qiita.com/sonots/items/979c8b0810ccffc47317)

## Gemfile
```Gemfile:.src/bundle_dir/Gemfile
source 'https://rubygems.org/'
gem 'embulk'

gem 'embulk-input-mysql'
gem 'embulk-output-mysql'
gem 'gimei'
```

入力も出力もRDS for Mysqlのためinとoutに`embulk-input-mysql`と`embulk-output-mysql`を使います
gimeiは自作のマスキングfilterプラグインのために使っています

# Dockerfile
```Dockerfile:Dockerfile
FROM openjdk:8-jre-alpine

RUN apk add --no-cache libc6-compat python py2-pip coreutils tzdata && \
  cp /usr/share/zoneinfo/Asia/Tokyo /etc/localtime && \
  echo "Asia/Tokyo" > /etc/timezone

RUN wget -q https://dl.embulk.org/embulk-0.9.18.jar -O /usr/local/bin/embulk && \
  chmod +x /usr/local/bin/embulk
RUN apk --update add libc6-compat

# ./src/bundle_dir/Gemfileでインストールするため不要
# RUN /usr/local/bin/embulk gem install embulk-input-mysql && \
#   /usr/local/bin/embulk gem install embulk-output-bigquery

RUN mkdir /workspace
COPY /src /workspace
WORKDIR /workspace/bundle_dir
RUN /usr/local/bin/embulk bundle

WORKDIR /workspace
COPY entrypoint.sh /usr/bin/entrypoint.sh
RUN chmod +x /usr/bin/entrypoint.sh
CMD ["entrypoint.sh"]
```

`./src`と`./entrypoint.sh`をCOPYし、CMDで最後にentrypoint.shを実行します
DBは外部のものに接続するため作りません。そういったシンプルな構造のため、今回はdocker-composeは使いませんでした。

参考： [Fargate環境でembulkを使ってMySQLからBigQueryへのマスタデータ転送](https://labs.gree.jp/blog/2019/03/17834/)

# 開発時のDockerコンテナ立ち上げ
```terminal:terminal
$ docker build -t embulk .
$ docker run --mount type=bind,source="$(pwd)"/src,target=/workspace --mount type=bind,source="$(pwd)"/entrypoint.sh,target=/usr/bin/entrypoint.sh -it embulk /bin/sh
```

`src`と`entrypoint.sh`をdocker run時にボリュームに指定し、コンテナ内のシェルに入ります。

# config.yml
embulkの設定ファイルを作ります
テーブルごとに1ファイル作る必要があるので、inとoutに関してはliquidを使用して`./src/config_commons`直下に切り分けます

参考： [Embulkの設定ファイルをincludeで共有化する方法](https://qiita.com/hiroysato/items/e7987e71a8112584b748)

## in_config
```yml:./src/config_commons/_in_config.yml.liquid
in:
  type: mysql
  host: {{ env.SOURCE_DB_HOST }}
  user: {{ env.SOURCE_DB_USER }}
  password: {{ env.SOURCE_DB_PASSWORD }}
  database: {{ env.DB_NAME }}
  table: {{ table }}
  options: {useLegacyDatetimeCode: false, serverTimezone: UTC}
  select: {{ select }} 
  where: {{ where }}
```

liquidの記法で`{{ }}`にて変数を展開することができます。また`env.*`で環境変数を呼び出すことができます

参考：
（変数） https://www.embulk.org/docs/built-in.html#using-variables
（Configuration） https://github.com/embulk/embulk-input-jdbc/tree/master/embulk-input-mysql
（options）[embulkでMySQLにコピーする時にハマったこと](https://qiita.com/s5601026/items/61c3a2f24d78ea296680#datetime%E5%9E%8B%E3%82%AB%E3%83%A9%E3%83%A0%E3%81%AE%E6%99%82%E9%96%93%E3%81%8C9%E6%99%82%E9%96%93%E3%82%BA%E3%83%AC%E3%81%9F)
 
## out_config
```yml:./src/config_commons/_out_config.yml.liquid
out:
  type: mysql
  host: {{ env.TARGET_DB_HOST }}
  user: {{ env.TARGET_DB_USER }}
  password: {{ env.TARGET_DB_PASSWORD }}
  database: {{ env.DB_NAME }}
  table: {{ table }}
  options: {useLegacyDatetimeCode: false, serverTimezone: UTC, characterEncoding: UTF-8}
  mode: truncate_insert
  before_load: {{ before_load }}
  after_load: {{ after_load }}
```

今回modeはtruncate_insertにしました。replaceにするとテーブルが作り直されてしまい、インデックスなども消えてしまう場合があるため注意が必要です。

参考： https://github.com/embulk/embulk-output-jdbc/tree/master/embulk-output-mysql

## user_config
```yaml:./src/user_config.yml.liquid
{% include 'config_commons/in_config',
    table: 'users',
    select: 'id, name, name_alphabet, age, email, user_type, created_at, updated_at' %}
{% include 'config_commons/out_config',
    table: 'users_copy' %}
```

次にFilterプラグインを自作していきます

# Filterプラグイン自作
`embulk emkbundle`で作った`bundle_dir`内の`bundle_dir/embulk/filter`内にプラグインを自作していきます（今回は公開までは行きません）

## ディレクトリ構造
```
./src/bundle_dir/embulk
├── filter
│   ├── example.rb
│   └── faker.rb
├── input
│   └── example.rb
└── output
    └── example.rb
```

## faker.rb
example.rbを参考にしながら書いていきます（公開する予定がなかったのでかなり雑に書いてますがお許しを）

```ruby:./src/bundle_dir/embulk/filter/faker.rb
require 'gimei'
require 'romaji' # gimeiの依存ライブラリ

module Embulk
  module Filter

    class FakerFilterPlugin < FilterPlugin
      # filter plugin file name must be: embulk/filter/<name>.rb
      Plugin.register_filter('faker', self)

      def self.transaction(config, in_schema, &control)
        task = {
          'columns_params' => config.param('mask', :array)
        }
        out_columns = in_schema

        puts "Faker filter started."
        yield(task, out_columns)
        puts "Faker filter finished."
      end

      def initialize(task, in_schema, out_schema, page_builder)
        super
        @columns_params = task['columns_params']
      end

      def close
      end

      def add(page)
        page.each do |record|
          fakes = make_fakes
          @columns_params.each do |column|
            select_column = page.schema.select{|c| c.name == column['name'] }
            next unless select_column.size == 1
            embulk_col = select_column.first

            record[embulk_col.index] = make_val(params, fakes)
          end

          @page_builder.add(record)
        end
      end

      def finish
        @page_builder.finish
      end

      def make_fakes
        gimei = Gimei.name
        return {
          'name_kanji' => gimei.kanji,
          'name_romaji' => (Romaji.kana2romaji gimei.katakana).split.map(&:capitalize).join(' ')
        }
      end

      def make_val(params, fakes)
        return params['value'] if params['type'] == 'replace'
        return fakes[params['type']]
      end
    end

  end
end
```

今回は日本人名のアルファベットを入れたかったので、カタカナもいけるGimeiを使っていますが、faker-rubyを使えば住所などいろいろ対応できると思います

## user_config.yml（追記）
```yaml:./src/user_config.yml
in:（略）
filters:
- type: faker
  mask:
  - {name: name, type: name_kanji}
  - {name: name_alphabet, type: name_romaji}
  - {name: email, type: replace, value: '*****@example.com'}
out:（略）
```

# entrypoint.sh

```sh:entrypoint.sh
#!/bin/sh
process_tables() {
    ls /workspace/*.yml.liquid | xargs -n1 java -jar /usr/local/bin/embulk run 2>&1 -I /workspace/bundle_dir -b /workspace/bundle_dir
}

init() {
    process_tables
}
init "$@"
```

これで`./src`以下に`*.yml.liquid`でYAMLファイルを作ってあげれば、その定義に基づいて複数テーブルをコピーできます

ちなみにmkbundleでディレクトリを作っているので、そのディレクトリを指定してあげなくてはならないのですが、bオプションのみでは`embulk-mysql-input`など公開gemが使えず、Iオプションのみでは自作の`faker`が使えないという事態に陥ったので、両方指定して無理やりやってます。
手動ではbオプションのみ`embulk run -b /workspace/bundle_dir`で実行できたのですが、シェルスクリプトでやらせると両方ないとできませんでした。

参考： Fargate環境でembulkを使ってMySQLからBigQueryへのマスタデータ転送](https://labs.gree.jp/blog/2019/03/17834/)

# 実行
これで環境変数を設定してあげて、シェルでentrypoint.shを実行してあげればできると思います
（めんどくさかったのでDocker shell内でmode変更していますが、ユーザーとかちゃんと設定してあげればなしでもいけるっぽいです）

```terminal:DockerShell
# chmod +x /usr/bin/entrypoint.sh
# entrypoin.sh
```