mysql間のデータ転送にEmbulk使ってみた

この記事は, [MicroAd Advent Calendar](https://qiita.com/advent-calendar/2020/microad) 2020の19日目の記事です.

Embulkを初めて使う人が試しに使ってみただけの記事です.
##はじめに
MySQLからBigQueryにデータ転送する機会があり, いい方法ないかと調べていたらEmbulkというものに出会った. 
動作確認のためにdocker-composeでMySQLのコンテナたててMySQL間のデータ転送したりしていたのでまとめておく.

とても便利だった

## Embulkとは
公式ページ(https://www.embulk.org/) から引用
>Embulk is a bulk data loader. It helps data transfer between types of databases, storages, file formats, cloud services, and else.

![スクリーンショット 2020-12-07 20.57.52.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/884891/1753aa7c-60c1-18a1-9b68-1ceda05b1f57.png)

Embulkはデータベース、ストレージ、ファイルフォーマット、クラウドサービスなどの種類の間のデータ転送を支援するバルクデータローダーでMySQLだけでなく, BigQueryなどにデータ転送が可能.

## 使ってみる
今回はdocker-composeでEmbulkを実行するコンテナとMySQLのコンテナをつないで動かしてみた.
ディレクトリ構造は以下の感じ.

.
├── db1/
│   &nbsp;&nbsp;&nbsp;&nbsp;├── Dockerfile
│   &nbsp;&nbsp;&nbsp;&nbsp;└── initdb.d/
│       &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;├── schema.sql
│       &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;└── testdata.sql
├── db2/
│   &nbsp;&nbsp;&nbsp;&nbsp;├── Dockerfile
│   &nbsp;&nbsp;&nbsp;&nbsp;└── initdb.d/
│       &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;├── schema.sql
│       &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;└── testdata.sql
├── docker-compose.yml
└── embulk/
    &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;├── Dockerfile
    &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;├── config.yml
    &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;└── seed.yml

MySQLのDockerfileは以下 (db2/Dockerfileも同様)

```docker:db1/Dockerfile
FROM mysql:5.7
RUN touch /var/log/mysql/mysqld.log # 指定の場所にログを記録するファイルを作る
```

Embulkはjava8が必要らしいからjava:8のイメージをベースにEmbulkをインストールする

```docker:embulk/Dockerfile
FROM java:8

# embulkインストール
RUN curl --create-dirs -o /usr/local/bin/embulk -L "https://dl.embulk.org/embulk-latest.jar" &&\
    chmod +x /usr/local/bin/embulk
# 各種プラグインインストール
RUN embulk gem install embulk-input-mysql
RUN embulk gem install embulk-output-mysql
RUN embulk gem install embulk-filter-concat
RUN embulk gem install embulk-filter-column
# RUN embulk gem install embulk-output-bigquery (BigQueyに転送する場合はこちらのプラグインが必要)
```

```yml:docker-compose.yml
version: '3.3' # docker-composeのversionを指定
services:
  db1:
    build: ./db1       
    restart: always
    environment:
      MYSQL_DATABASE: sample_db
      MYSQL_USER: user
      MYSQL_PASSWORD: password
      MYSQL_ROOT_PASSWORD: rootpassword
    ports:
      - "3306:3306"
    volumes:
      - ./db1/initdb.d:/docker-entrypoint-initdb.d
  db2:
    build: ./db2      
    restart: always
    environment:
      MYSQL_DATABASE: sample_db
      MYSQL_USER: user
      MYSQL_PASSWORD: password
      MYSQL_ROOT_PASSWORD: rootpassword
    ports:
      - "3307:3307"
    volumes:
      - ./db2/initdb.d:/docker-entrypoint-initdb.d
  embulk:
    build: ./embulk
    volumes:
      - .:/workspace
    stdin_open: true
    tty: true
    working_dir: /workspace
```
テーブルは適当に初期値入れておく

```sql:db1/init.db.d/schema.sql
CREATE TABLE test (
    id INT NOT NULL AUTO_INCREMENT,
    name VARCHAR(32) NOT NULL,
    email VARCHAR(32) NOT NULL,
    PRIMARY KEY (id)
);
```
```sql:db1/init.db.d/schema.sql
INSERT INTO test (id,name,email) 
VALUES (1, 'aaa','aaa@mail.co.jp'), (2, 'bbb','bbb@mail.co.jp'), (3, 'ccc','ccc@mail.co.jp');
```

とりあえずコンテナ立ち上げる

```
$ docker-compose up -d --build
```
確認

```
$ docker-compose ps

        Name                     Command             State                      Ports                   
--------------------------------------------------------------------------------------------------------
embulk-test_db1_1      docker-entrypoint.sh mysqld   Up      0.0.0.0:3306->3306/tcp, 33060/tcp          
embulk-test_db2_1      docker-entrypoint.sh mysqld   Up      3306/tcp, 33060/tcp, 0.0.0.0:3307->3307/tcp
embulk-test_embulk_1   /bin/bash                     Up  
```

embulk-test_db1_1のMySQLにレコードが入っていることを確認

```
$ docker exec -it embulk-test_db1_1 /bin/bash
$ mysql -u user -p
mysql> select * from test;

+----+------+----------------+
| id | name | email          |
+----+------+----------------+
|  1 | aaa  | aaa@mail.co.jp |
|  2 | bbb  | bbb@mail.co.jp |
|  3 | ccc  | ccc@mail.co.jp |
+----+------+----------------+
3 rows in set (0.00 sec)
```

このテーブルをembulk-test_db2_1のコンテナで動いているMySQLに転送したい

### Embulk使って転送
embulkの設定ファイル記述. ymlで記述する.
MySQLの場合は以下の感じ. 
今回は入力と出力が両方ともMySQLだから, それぞれ必要なプラグインは
`embulk-input-mysql`と`embulk-output-mysql`(予めインストールしてある)

```yml:embulk/config.yml
in:
  type: mysql
  host: embulk-test_db1_1
  user: user
  password: password
  database: sample_db
  table: test
  select: id, name, email

out:
  type: mysql
  host: embulk-test_db2_1
  user: user
  password: password
  database: sample_db
  table: test
  mode: replace
```
(tableとselectを指定しているが, 直接クエリを書くやり方もある)

Embulkを実行する前に転送される内容をpreviewできる.

```
$ embulk preview config.yml 

2020-12-13 07:53:45.496 +0000: Embulk v0.9.23
2020-12-13 07:53:46.237 +0000 [WARN] (main): DEPRECATION: JRuby org.jruby.embed.ScriptingContainer is directly injected.
2020-12-13 07:53:47.793 +0000 [INFO] (main): Gem's home and path are set by default: "/root/.embulk/lib/gems"
2020-12-13 07:53:48.353 +0000 [INFO] (main): Started Embulk v0.9.23
2020-12-13 07:53:48.453 +0000 [INFO] (0001:preview): Loaded plugin embulk-input-mysql (0.11.1)
2020-12-13 07:53:48.476 +0000 [INFO] (0001:preview): Loaded plugin embulk-filter-concat (0.1.0)
2020-12-13 07:53:48.516 +0000 [INFO] (0001:preview): Loaded plugin embulk-filter-column (0.7.1)
2020-12-13 07:53:48.538 +0000 [INFO] (0001:preview): JDBC Driver = /root/.embulk/lib/gems/gems/embulk-input-mysql-0.11.1-java/default_jdbc_driver/mysql-connector-java-5.1.44.jar
2020-12-13 07:53:48.546 +0000 [INFO] (0001:preview): Fetch size is 10000. Using server-side prepared statement.
2020-12-13 07:53:48.548 +0000 [INFO] (0001:preview): Connecting to jdbc:mysql://embulk-test_db1_1:3306/sample_db options {useCompression=true, socketTimeout=1800000, useSSL=false, user=user, useLegacyDatetimeCode=false, tcpKeepAlive=true, useCursorFetch=true, connectTimeout=300000, password=***, zeroDateTimeBehavior=convertToNull}
2020-12-13 07:53:48.727 +0000 [INFO] (0001:preview): Using JDBC Driver mysql-connector-java-5.1.44 ( Revision: b3cda4f864902ffdde495b9df93937c3e20009be )
2020-12-13 07:53:48.727 +0000 [WARN] (0001:preview): embulk-input-mysql 0.9.0 upgraded the bundled MySQL Connector/J version from 5.1.34 to 5.1.44 .
2020-12-13 07:53:48.728 +0000 [WARN] (0001:preview): And set useLegacyDatetimeCode=false by default in order to get correct datetime value when the server timezone and the client timezone are different.
2020-12-13 07:53:48.728 +0000 [WARN] (0001:preview): Set useLegacyDatetimeCode=true if you need to get datetime value same as older embulk-input-mysql.
2020-12-13 07:53:48.833 +0000 [INFO] (0001:preview): Fetch size is 10000. Using server-side prepared statement.
2020-12-13 07:53:48.833 +0000 [INFO] (0001:preview): Connecting to jdbc:mysql://embulk-test_db1_1:3306/sample_db options {useCompression=true, socketTimeout=1800000, useSSL=false, user=user, useLegacyDatetimeCode=false, tcpKeepAlive=true, useCursorFetch=true, connectTimeout=300000, password=***, zeroDateTimeBehavior=convertToNull}
2020-12-13 07:53:48.841 +0000 [INFO] (0001:preview): SQL: SELECT id, dt, hour, created_at FROM `test`
2020-12-13 07:53:48.844 +0000 [INFO] (0001:preview): > 0.00 seconds
+--------+-------------+----------------+
| id:int | name:string | email:string   |
+--------+-------------+----------------+
|     1  |        aaa  | aaa@mail.co.jp |
|     2  |        bbb  | bbb@mail.co.jp |
|     3  |        ccc  | ccc@mail.co.jp |
+--------+-------------+----------------+
```
ここで問題があれば, config.ymlを修正して再確認する. 

問題がなさそうなら以下のコマンドでEmbulk実行

```
$ embulk run config.yml
```

embulk-test_db2_1に転送されてるか確認

```
mysql> select * from test;

+----+------+----------------+
| id | name | email          |
+----+------+----------------+
|  1 | aaa  | aaa@mail.co.jp |
|  2 | bbb  | bbb@mail.co.jp |
|  3 | ccc  | ccc@mail.co.jp |
+----+------+----------------+
3 rows in set (0.00 sec)
```
バッチリ!



## filter機能

filterを使えば, カラムの追加や型キャストなどいろいろできる.
例えば特定のカラムを結合したい場合は`embulk-filter-concat`を使って設定ファイルを以下のように変更する

```yml:embulk/config.yml
in:
  type: mysql
  host: embulk-test_db1_1
  user: user
  password: password
  database: sample_db
  table: test
  select: id, name, email

filters:
- type: concat
  name: namemail
  columns:
  - {name: name}
  - {name: email}

out:
  type: mysql
  host: embulk-test_db2_1
  user: user
  password: password
  database: sample_db
  table: test
  mode: replace
```

上の例では, nameとemailを連結して新たにnamemailというカラムを作っている. 

```
$ embulk preview config.yml 

2020-12-13 07:53:45.496 +0000: Embulk v0.9.23
2020-12-13 07:53:46.237 +0000 [WARN] (main): DEPRECATION: JRuby org.jruby.embed.ScriptingContainer is directly 
....

+--------+-------------+----------------+--------------------+
| id:int | name:string | email:string   |  namemail:string   |
+--------+-------------+----------------+--------------------+
|     1  |        aaa  | aaa@mail.co.jp | aaa aaa@mail.co.jp |
|     2  |        bbb  | bbb@mail.co.jp | bbb bbb@mail.co.jp |
|     3  |        ccc  | ccc@mail.co.jp | ccc ccc@mail.co.jp |
+--------+-------------+----------------+--------------------+
```

他にもたくさんプラグインがあるが, こちら[Embulk(エンバルク)プラグインのまとめ](https://qiita.com/hiroysato/items/da45e52fb79c39547f69) に細かくまとめられているので, ここでは省略する.

## まとめ

今回はmysql間でのデータ転送を試しただけだったが, 異種間のデータ転送もお手軽にできてとても便利だった.


**参考**
[Embulk](https://www.embulk.org/)
[Embulk(エンバルク)プラグインのまとめ](https://qiita.com/hiroysato/items/da45e52fb79c39547f69)