Embulkでバイナリのデータがコピーできない

Question

前提・実現したいこと

embulkでRDStoRDSでコピーをしたいのですが、バイナリデータがサポートされていていないようで、エラーになります。
エラーメッセージの通り、columnA: {value_type: string}を追加しても解消されませんでした。

何か回避方法をご存知であれば、教えて頂けると助かります。。

発生している問題・エラーメッセージ

Caused by: java.lang.UnsupportedOperationException: Unsupported type MEDIUMBLOB (sqlType=-4) of 'DATA' column. Please add 'columnA: {value_type: string}' to 'column_options: {...}' option to convert the values to strings, or exclude the column from 'select:' option

コード

_input.yml.liquid

in:
  type: mysql
  host: {{ env.IN_MYSQL_HOST }}
  user: {{ env.IN_MYSQL_USER }}
  password: {{ env.IN_MYSQL_PASSWORD }}
  database: dbA
  select: "*"
  options: {useLegacyDatetimeCode: false, serverTimezone: Asia/Tokyo}

_output.yml.liquid

out:
  type: mysql
  mode: replace
  host: {{ env.OUT_MYSQL_HOST }}
  user: {{ env.OUT_MYSQL_USER }}
  password: {{ env.OUT_MYSQL_PASSWORD }}
  database: dbA
  select: "*"

config.yml.liquid

{% include 'input' %}
  table: tableA
{% include 'output' %}
  table: tableA
  column_options:
    DATA: {value_type: string}

テーブル定義

ID:INT
DATA:MEDIUMBLOB
CREATED_DATE:DATETIME
UPDATED_DATE:DATETIME

バージョンなど

Amazon linux2
Embulk v0.9.23
embulk-input-mysql 0.10.1
embulk-output-mysql 0.8.7

2021/06/16 追記
下記のように修正し、embulkを実行したところ、エラーになりました。

in:
  type: mysql
  user: user
  password: password
  database: dbA
  table: tableA
  host: host
  select: id,TO_BASE64(DATA) as DATA
out:
  type: mysql
  user: user
  password: password
  database: dbA
  table: tableA
  column_options:
    DATA:
      type: longtext
  host: host
  mode: replace
  after_load: insert into tableA(id,DATA) select id,FROM_BASE64(DATA) from tableA;

Comments

[y_waiwai]

コードを提示しましょう

[koichi88]

すみません、情報が不足しておりました。
コードを追加させて頂きました。

Answer 1

embulkはバイナリはそのまま取り扱いできないので、base64などascii表現できる文字列に変換して取り出し、インサートする時にデコードするというような方法が必要な気がします。

取り出し

取り出しは以下のようにTO_BASE64を使ってbase64エンコードする方法が使えると思います。

create table strtest(str1 varbinary(20), str2 char(5) character set utf8);
insert into strtest values('空', '空');

in:
  type: mysql
  user: user
  password: password
  database: database
  table: strtest
  host: host
  select: TO_BASE64(str1),str2
out:
  type: stdout

これで、

56m6,空

という結果が出てきてecho "56m6" | openssl base64 -dでバイナリに格納した「空」がデコードできます。

書き込み

書き込みは、ちょっと難しく、replaceであれば
別テーブルにbase64のまま書き込みしていいて、embulk-output-mysqlのafter_loadオプションを使ってsqlでテーブルの洗い換えをする方法が一案かなと思います。

何か思いついたら補足しますが、インサート時にBase64データをデコードすることはできなかったような気がします。

その他

ご質問の際に以下の情報を書いていただけるとより回答しやすいです。

• Embulkのバージョン
• ご利用のOS
• プラグインのバージョン
• 設定ファイル(機密情報等はマスク)
• テーブル定義(サンプル等)

2021/06/15 追記

できそうなのはこんな感じです。

create table imagetest(id int not null primary key, image MEDIUMBLOB not null);
insert into  imagetest(id,image) values(1,LOAD_FILE('/path/to/image'))；
create table imagedst(id int not null primary key, image MEDIUMBLOB not null);

in:
  type: mysql
  user: user
  password: password
  database: embulk_test
  table: imagetest
  host: host
  select: id,TO_BASE64(image) as image
out:
  type: mysql
  user: user
  password: password
  database: embulk_test
  table: imagetemp
  column_options:
    image:
      type: longtext
  host: host
  mode: replace
  # ここは入れ替えするか挿入するか等によって修正
  after_load: insert into imagedst(id,image)  select id,FROM_BASE64(image) from imagetemp;

2021/06/17 追記

複数SQLの件は、ストアドプロシージャーが良いように思います。

delimiter //
create PROCEDURE sync_image_table ()
begin
  delete from imagedst;
  insert into imagedst(id,image)  select id,FROM_BASE64(image) from imagetemp;
end//

  after_load: call sync_image_table()

Comments

[koichi88]

回答ありがとうございます！！
追加で情報を追加させて頂きました。

[koichi88]

下記のように修正し、embulkを実行しました。

```
in:
select: "*"
↓
select: ID,TO_BASE64(DATA),CREATED_DATE,UPDATED_DATE
```

下記のようなエラーが表示されました。
org.embulk.exec.PartialExecutionException: java.lang.RuntimeException: java.sql.BatchUpdateException: Data truncation: Data too long for column 'TO_BASE64(DATA)' at row 2

DATAカラムは画像データのため、サイズが大きいのが、原因かと思います。。

[hiroysato]

画像ファイルは何MBぐらいでしょうか？
手元で900KBぐらいのイメージだったらうまく動きました。
バイナリを転送する場合は、他のツールの方が良いかもしれませんね。

[hiroysato]

これout側のエラーですかね？私の環境(type: stdout)では、12MBのイメージでもエラーは出ませんでした。
BASE64は通常のファイルの133%大きくなるのでそれが格納できるカラムじゃないとダメですね。

[hiroysato]

補足を書きました。

[koichi88]

githubも含め、回答ありがとうございます。
確認してみます。

[koichi88]

after_loadの箇所で下記のようにエラーになりました。
テーブルの作成、該当カラムのデータが挿入されているのですが、データによってはエラーが出るようです。

```
SQL: insert into tableA(ID,DATA) select ID,FROM_BASE64(DATA) from tableA;
2021-06-16 01:33:22.891 +0900 [ERROR] (0001:transaction): Operation failed (1366:HY000)
2021-06-16 01:33:22.912 +0900 [INFO] (0001:cleanup): Connecting to jdbc:mysql://db.ap-northeast-1.rds.amazonaws.com:3306/dbA options {user=user, password=***, tcpKeepAlive=true, useSSL=false, useCompression=true, rewriteBatchedStatements=true, connectTimeout=300000, socketTimeout=1800000}
2021-06-16 01:33:22.922 +0900 [INFO] (0001:cleanup): TransactionIsolation=repeatable_read
2021-06-16 01:33:22.922 +0900 [INFO] (0001:cleanup): SQL: DROP TABLE IF EXISTS `tableA_0000017a107dbe6b_embulk`
2021-06-16 01:33:22.925 +0900 [INFO] (0001:cleanup): > 0.00 seconds
org.embulk.exec.PartialExecutionException: java.lang.RuntimeException: java.sql.SQLException: Incorrect string value: '\xFF\xD8\xFF\xE0\x00\x10...' for column 'DATA' at row 17517
```

[koichi88]

out;のDATAカラムがlongtextになっているので、それをコピー元と同じようにMEDIUMBLOBにすれば、解決できそうです。

[hiroysato]

insert into tableA(ID,DATA) select ID,FROM_BASE64(DATA) from tableA;

Base64エンコードしたテーブルとデコードして挿入する先のテーブルが同じになっていませんか？

[koichi88]

同期先、同期元は同じテーブル名を利用しているので、同じになっています。
imagedstとimagetempは別の名前にする必要があるということでしょうか。

[hiroysato]

よくわからないですが、FROM_BASE64は、Base64エンコードしたカラムに対して実行するもので、バイナリのカラムにFROM_BASE64をしたら、エラーがでるのは当然だと思います。
設定ファイルをみていただければわかるように、embulk-output-mysqlは imagetemp というテーブルにBase64エンコードしたまま入れて、after_loadで、ターゲットのテーブルに入れています。

読み出しと書き出しが一緒ならこうなると思います。

* embulk-input-mysqlで読み出すテーブル: tableA
* embulk-output-mysqlで読み出すテーブル: tableTMP
* embulk-output-mysqlのafter_loadのFROM: tableTMP
* embulk-output-mysqlのafter_loadのINSERT INTO: tableA

[koichi88]

回答ありがとうございます。
コピーには成功したのですが、２回目を実行した場合、重複してデータが挿入されている状態になります。
重複しているデータを無くすようにするにはafter_loadの最初にテーブル内のレコードを全て削除する方法で対応するのでしょうか。

```
after_load: DELETE FROM imagedst; insert into imagedst(id,image) select id,FROM_BASE64(image) from imagetemp;
```

上記の記述だとエラーになる状態です。
```
Caused by: java.lang.RuntimeException: com.mysql.jdbc.exceptions.jdbc4.MySQLSyntaxErrorException: You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'insert into image
```

[hiroysato]

はい、全部を消してから挿入、存在しないIDだけ消すて挿入・マージ、テーブル名を変更等のクエリになると思います。

[koichi88]

回答ありがとうございます。
after_loadで複数クエリを書く方法がまだわかっていないのですが、この度はご対応ありがとうございました。

[koichi88]

issueとして新規作成したので、大変恐縮ですが、何かご存知であれば、ご回答頂けると幸いです。
https://github.com/embulk/embulk-input-jdbc/issues/219

[koichi88]

ストアドプロシージャーの件、ありがとうございます。
試してみます！