跳转至

mysql 迁移，指定某一时间的审计日志进行回放。

前言

这个软件我们在从腾讯迁移数据库到阿里的时候，阿里的同学给我们用过，所以觉得很好用，然后我就去github查找了一下，阿里居然开源了。主要用的是frodo， 开源版本的跟他们提供给客户的参数上和版本上也是有区别的，但是开源也够用了。

我的源数据是从腾讯云的审计日志里面提取的，并且转换成frodo能识别的日志。

1、配置阿里的maven仓库

增加一个配置就行,原文 

<mirror>
  <id>aliyunmaven</id>
  <mirrorOf>*</mirrorOf>
  <name>阿里云公共仓库</name>
  <url>https://maven.aliyun.com/repository/public</url>
</mirror>

2、编译安装frodo

参考github中的readme

3、解析腾讯云审计日志

from datetime import datetime
import pandas as pd
import json

input_filename = '123.csv'
output_filename = 'sql.txt'
columns_to_read = ['Sql', 'User', 'ExecTime', 'ThreadId', 'Timestamp', 'DBName']
date_format = "%Y-%m-%d %H:%M:%S"

df = pd.read_csv(input_filename, usecols=columns_to_read)

for index, row in df.iterrows():
   dt_object = datetime.strptime(row['Timestamp'], date_format)
   timestamp_in_seconds = dt_object.timestamp()
   timestamp_in_microseconds = int(timestamp_in_seconds * 1_000_000)

   _dict={
     "schema":row['DBName'],
     "session":row['ThreadId'],
     "execTime":row['ExecTime'],
     "startTime":timestamp_in_microseconds,
     "convertSqlText":row['Sql'],
     "user":row['User']
   }
   print(json.dumps(_dict))

   #with open(output_filename, 'w+') as json_file:
   # json.dump(json.dumps(_dict), json_file, indent=4)
最终生成的报告中，模板的成功率越高越好，低的话，回放出来的数据库监控数据不一定准确。

  • convertSqlText中的内容不要为NaN
  • 包含’/‘ 可能会执行失败

生成的数据格式： 

{"schema": "dbname", "session": 6666, "execTime": 83, "startTime": 1721666395000000, "convertSqlText": "SELECT  * FROM test WHERE  id = '6666' LIMIT 1", "user": "test"}

4、执行frodo

java -Xmx8G -Xms1G -jar frodo.jar --file=/root/789.json --source-db=mysql --replay-to=mysql --port=3306 --host=xxxxx --username=xxx --password='xxxx' --concurrency=15 --time=2400 --task=0723-00-40 --schema-map=xxx --log-level=info --rate-factor=1 --database=xxx --filter=DQL

本文阅读量  次

评论