embulkの増分バルクインサート処理の仕組みについて

Question

embulkの増分バルクインサート処理を使って、Oracle DBのデータをBigQueryへデータ連携させます。

増分バルクインサート処理の仕組みについて確認したいです。

※認識している増分バルクインサート処理のロジック

➀Embulk実行
➁OracleへJDBC接続
➂BigQueryに中間テーブルを生成
➃インサート対象のレコード検索
➄embulk内部ファイルへ書込み
➅embulk内部ファイルを中間テーブルへコピー
➆中間テーブルからBigQueryのテーブルへバルクインサート
➇中間テーブル削除
➈embulk内部ファイル削除
➉コミット
⒒結果ファイルへlast_record書き込み

上記ロジックであると認識していまして、

➃インサート対象のレコード検索

➃の処理について、実際DBに対してどのような検索をしているのか？

上記１１の処理で、結果ファイルにlast_record(タイムスタンプ)を出力し、次回インサート処理時には結果ファイルを参照して、前回インサートしたレコード分はスキップする。

との情報を得ているのですが、embulkの内部DB？があって情報整理しているのか？
教えて頂きたいです。

Answer 1

-c diff.yml で指定したファイルにlast_recordの情報が記録され次回実行時にwhere句に
last_record > 'last_record'の値
という文が追加されるだけです。out: stdoutとかにして実行してみるのがわかりやすいと思います。

Comments

[raccoondog]

ご回答有難うございます。
まだ、理解出来ておりませんでして、再確認させてください。

embulkの結果ファイルにlast_recordの値(yyyy_mm_dd hh:mm:ss)が出力されていて、次回実行時にはinput対象のテーブルに対して、以下のSQLを使い

select * from table_name where column_name > (yyyy_mm_dd hh:mm:ss) order by column_name;

増分バルクインサート処理される。

この時のタイムスタンプ情報と、特定のカラム値の情報をどのように紐づけしているのか知りたいです。

例：table

ID NUM STR DT
--- ------ ------ -----
11 1234 abc 1210
12 5678 def 1211
13 9123 ghi 1212 ←　ここのレコード迄が前回インサート分

例:diff.yml

2019-02-13T09:00:00.00000

Answer 2

3回目が以下のデータで

ID NUM STR DT 
--- ------ ------ -----
11 1234 abc 1210 
12 5678 def 1211 # <-- 初回取得レコード
13 9123 ghi 1212
14 4567 ghi 1213 # <-- 2回目取得レコード
15 8912 ghi 1214
16 3456 jkl 1215 # <-- 3回目取得レコード

diff.ymlが以下のようになっており

{ DT: 1215}

4回目が以下のデータで、

ID NUM STR DT 
--- ------ ------ -----
11 1234 abc 1210 
12 5678 def 1211 # <-- 初回取得レコード
13 9123 ghi 1212
14 4567 ghi 1213 # <-- 2回目取得レコード
15 8912 ghi 1214
16 3456 jkl 1215 # <-- 3回目取得レコード
17 7891 mno 1215 ☆また、1215と同じカラム値が連続していた場合

ID = 17のレコードが取得できるかというご質問であれば、DTを使っている場合はID=17は取得できません。
4回目の実行は以下のようになるからです。

select * from table where DT > 1215

incremental_columnで指定するカラムは後から挿入されるデータが必ず現在保存されているより大きい値であることを期待します。

数値の場合は大きい数字、時刻の場合はより新しい時刻です。

Comments

[raccoondog]

有難うございます。
大変助かりました。

Answer 3

現在うまく言っているのであれば、そちらの設定ファイルと実行時のログを見ればどのように紐付けされるのかわかるような気がします。
現在の設定と実行されているSQL文を確認してみてはいかがでしょうか？

私を含めて他の人も環境、設定ファイル、ログがないと具体的には回答できません。

以下のようになるはずです。現在うまく動作しているログがあると思いますのでそちらを確認することをお勧めします。

1回目

ID NUM STR DT 
--- ------ ------ -----
11 1234 abc 1210 
12 5678 def 1211

incremental: true
incremental_column: [DT]

embulk実行、以下のどちらかが実行されます。
(実行しないとわからないのでご自身で確認してみてください。)

select * from table where DT > 0

または

select * from table

作成されるdiff.yml

{ DT: 1211 }

2回目

ID NUM STR DT 
--- ------ ------ -----
11 1234 abc 1210 
12 5678 def 1211 # <-- 前回取得レコード
13 9123 ghi 1212 # <-- 追加レコード

select * from table where DT > 1211

Comments

[raccoondog]

有難うございます。ロジックの詳細がわかりました。
実行ログと、設定ファイル、テーブルの値を見比べて、上記回答の通りになっておりました。

[raccoondog]

すみません、追加で確認させてください。
例えば以下の状況の場合でも最終取得レコードを正常に判別できるのでしょうか？

ID NUM STR DT
--- ------ ------ -----
11 1234 abc 1210
12 5678 def 1211 # <-- 初回取得レコード
13 9123 ghi 1212
14 4567 ghi 1213 # <-- 2回目取得レコード
15 8912 ghi 1214
16 3456 jkl 1215 # <-- 3回目取得レコード
17 7891 mno 1215 ☆また、1215と同じカラム値が連続していた場合

Answer 4

こちらを参考に、
https://obel.hatenablog.jp/entry/20170302/1488383422

incremental: true
incremental_column: [DT]

とやってみるのはいかがですか？

前うまくできたと書いてありますが、前回となにが違うのかよくわかりません。
https://teratail.com/questions/160529

Comments

[raccoondog]

ご回答有難うございます。

設定につきましては、問題なく成功しております。

前回の質問にご回答頂いた内容で、何の問題もない状況です。

embulkの増分バルクインサート処理のロジックを詳細に知りたい状況となります。

といいますのも、1億レコード近くあるDBに、1日最大50万レコードがDBにインサートされるシステムになります。

頻繁にインサート処理がされるOracle DBに対して、embulkを使ってbigqueryへ増分バルクインサート処理を定期実行する運用で、ファイルに出力されるタイムスタンプ値のみで本当に問題ないのか？　とプロジェクト内で議題に上がっております。

そこで、ロジックの詳細を抑えておいて説明が出来るようにしたいです。