DataXを活用したMySQLデータベースの増分同期実装ガイド

DataXを使用したデータの全量同期は簡単ですが、増分同期はどう実装すればよいでしょうか。本記事ではその方法を詳しく解説します。

DataXのダウンロードとセットアップ（CentOSにJDKなどの実行環境が既にインストールされていることが前提）

wget http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/datax.tar.gz
tar -zxvf datax.tar.gz && mv datax /usr/local/
cd /usr/local/datax/bin/
python datax.py /usr/local/datax/job/job.json

エラーが発生した場合は、以下のブログを参考にしてください：https://www.cnblogs.com/juanxincai/p/16258154.html

問題解決策：

cd /usr/local/datax/plugin/reader
ll -a
[root@Data1 reader]# ll -a
total 76
drwxr-xr-x 21 502 games 4096 Feb 19 21:05 .
drwxr-xr-x  4 502 games   66 Feb 19 21:05 ..
drwxr-xr-x  3 502 games  224 Feb 19 21:05 cassandrareader
-rwxr-xr-x  1 502 games  212 Oct 12  2019 ._cassandrareader
....
._で始まるファイルを削除
rm -f ._*
cd /usr/local/datax/plugin/writer/
rm -f ._*

これで、正常に自己診断が完了したはずです。

同期のアプローチ：

Pythonを使用してデータソースのデータベーステーブルをクエリし、取得した最後のデータの時間をテキストファイルに保存します。次回の実行時にその時間を読み取り、cronタスクを組み合わせることで、時間間隔内のデータを定期的に同期します（増分同期）。

PyMySQLを使用するにはPython3へのアップグレードが必要です。CentOS7にはPython2が標準で搭載されています。インストール方法はこちらを参照：https://www.cnblogs.com/juanxincai/p/16280031.html

インストール後、再度自己診断を実行し、エラーが発生した場合はこちらを参照：https://www.cnblogs.com/juanxincai/p/16284779.html

次に、以下の4つの要素が必要です：

読み取りと書き込みを実行するmysqltomysql.json（ここではファイル名をnew.jsonとしています）。データソースのデータベーステーブル情報、読み取りフィールドなどの設定が含まれ、外部から2つの時間パラメータを受け取ります（タイムスタンプにフォーマット）。パスは/usr/local/datax/jobにあり、JSON形式は標準形式である必要があります。使用前にフォーマットに問題がないか確認してください。

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "username": "ユーザー名",
                        "password": "パスワード",
                        "where": "data_time >= FROM_UNIXTIME(${create_time}) and data_time < FROM_UNIXTIME(${end_time})",
                        "column": [
                            "id","data_time","name","age","insert_time"
                        ],
                        "connection": [
                            {
                                "table": [
                                    "テーブル名"
                                ],
                                "jdbcUrl": [
                                    "jdbc:mysql://データソースIP:3306/データベース名?useUnicode=true&characterEncoding=utf8"
                                ]
                            }
                        ]
                    }
                },
                "writer": {
                    "name": "mysqlwriter",
                    "parameter": {
                        "writeMode": "update",
                        "username": "ユーザー名",
                        "password": "パスワード",
                        "column": [
                            "id","data_time","name","age","insert_time"
                        ],
                        "connection": [
                            {
                                "jdbcUrl": "jdbc:mysql://ターゲットDBIP:3306/データベース名?useUnicode=true&characterEncoding=utf8",
                                "table": [
                                    "テーブル名"
                                ]
                            }
                        ]
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": 6
            }
        }
    }
}

定時実行するPythonスクリプト。データソースのデータベーステーブルから最後のデータの時間を取得し、テキストファイルに書き込み、datax.pyを実行して上記のnew.jsonを実行します（ファイル名：mysql2mysqlexecute.py）。

#!/usr/bin/env python3
# coding: utf-8
import subprocess as sp
import time, os, sys
import pymysql
import pickle

print ("実行を開始します")

configFilePath = sys.argv[1]
logFilePath = sys.argv[2]
lastDataTime=""


def save_variable(v, filename):
    f = open(filename, 'wb')
    pickle.dump(v, f)
    f.close()
    return filename

def load_variable(filename):
    ff = open(filename, 'rb')
    r = pickle.load(ff, encoding="UTF-8")
    ff.close()
    return r
	
startTime=load_variable('/usr/local/datax/job/tempTime.txt')  # 一時的な時間変数を保存するtxtファイル。エンコード形式に注意してください。開始時間を読み取ります

def do_sql(sql):
	db = pymysql.connect(host = 'データソースDBIP', port = 3306, user = 'ユーザー名', passwd = 'パスワード', db = 'データベース名')
# 接続を作成（データベースに接続）
	cursor = db.cursor()  # カーソルを作成
	cursor = db.cursor(cursor=pymysql.cursors.DictCursor)   # カーソル形式を辞書形式に設定。値の取得時に辞書形式で表示されます
	cursor.execute(sql) # SQL文を実行
	rs=cursor.fetchall() 
	#for r in rs: 
	#print (r)
	content=rs
	db.commit()  # 実行結果を保存するためにコミット
	cursor.close()   # カーソルを閉じる
	db.close()     # 接続を閉じる
	x = rs[0]['dataTime']
	return x;
	
print("startTime=",startTime) # フォーマットされた同期開始日を出力
startTimeArray = time.strptime(startTime, "%Y-%m-%d %H:%M:%S")
startTimeStamp = int(time.mktime(startTimeArray))

sql='SELECT CAST(data_time AS CHAR) as dataTime FROM テーブル名 ORDER BY data_time DESC LIMIT 1'	

lastDataTime = do_sql(sql)
print("endTime=",lastDataTime)
lastDataTimeArray = time.strptime(lastDataTime, "%Y-%m-%d %H:%M:%S")
lastDataTimeTimeStamp = int(time.mktime(lastDataTimeArray))

try:
	script2execute  = "/usr/bin/python3 /usr/local/datax/bin/datax.py %s -p \"-Dcreate_time=%s -Dend_time=%s\" >> %s"%(configFilePath,startTimeStamp,lastDataTimeTimeStamp,logFilePath)
	print("実行するスクリプト:",script2execute)
	os.system(script2execute)
	#sp.run(script2execute)
except IOError:
		print(IOError) 

print("スクリプトの実行が終了しました")

save_variable(lastDataTime,'/usr/local/datax/job/tempTime.txt') # 一時的な時間変数を保存し、次回の開始時間として使用

print("終了---")

定時同期タスクを実行するshスクリプト：（ファイル名:timeMission.sh）。ここで、new.jsonファイルの場所とログのパスが渡されていることがわかります。

#! /bin/bash
source /etc/profile

/usr/bin/python3 /usr/local/datax/job/mysql2mysqlexecute.py  '/usr/local/datax/job/new.json'  '/usr/local/datax/job/test_job.log'   '/usr/local/datax/job/test_job.record'

上記の読み取りと一時的な時間変数を保存するファイル名はtempTime.txtです。自分で作成してください。パスとエンコード形式に注意してください。

shを使用する場合は、スクリプトに実行権限を付与することを忘れないでください。

chmod +x ./xxx.sh

次に、cronを使用して定時タスクを設定します。問題があれば、以下を参照してください：https://www.cnblogs.com/juanxincai/p/15852374.html

crontab -e

SHELL=/bin/bash
 */5 * * * *  /usr/local/datax/job/timeMission.sh

:wqで保存して終了。この2行を追加すると、timeMission.shが5分ごとに一度実行され、5分ごとに同期が実行されます```

crontab -l

これで、定時タスクが書き込まれたことがわかります。

systemctl reload crond.service
systemctl restart crond.service

cronサービスを再起動して再読み込みし、タスク実行のログ出力を確認します。

tail -f /var/spool/mail/root

次に、dataxの実行ログを確認します。

 tail -200f /usr/local/datax/job/test_job.log

データが同期されたことがわかります。さらに最適化するには、dataxの公式ドキュメントを参照してください。セットアップ中に問題が発生しても恐れず、大胆に試行錯誤し、注意深くデバッグしてください。フォーマット、実行権限、ファイルのエンコードなどが原因で成功しない場合があります。皆さんが一度で同期に成功することを願っています。

タグ: DataX MySQL 増分同期定時タスク PyMySQL

10月11日 04:34 投稿

異端開発室

DataXを活用したMySQLデータベースの増分同期実装ガイド

ホットタグ