MySQL数据迁移TcaplusDB实践

CREATE TABLE `test` (
  `player_id` bigint(20) NOT NULL,
  `player_name` varchar(128) DEFAULT NULL,
  `player_info` varbinary(2048) DEFAULT NULL,
  PRIMARY KEY (`player_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4

syntax = "proto3";               // Specify the version of the protocol buffers language

import "tcaplusservice.optionv1.proto"; // Use the public definitions of TcaplusDB by importing them.

message test {  // Define a TcaplusDB table with message

    // Specify the primary keys with the option tcaplusservice.tcaplus_primary_key
    // The primary key of a TcaplusDB table has a limit of 4 fields
    option(tcaplusservice.tcaplus_primary_key) = "player_id";

    // primary key fields
    int64 player_id = 1;
    //ordinary fields
    string player_name = 2;
    info player_info = 3;

}

message info {
    string player_email = 1;
    string player_phone = 2;
}

//替换TOPIC为申请的topic名，这里为tw_migrate
final String TOPIC = "tw_migrate";
Properties props = new Properties();

//替换CKafka实例的连接信息，进Ckafka控制查看实例信息中的内网IP与端口
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.17.16.6:9092");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
final Producer<String, String> producer = new KafkaProducer<String, String>(props);

//替换SecretId为自己腾讯云账号下申请的密钥信息
context.setSecretId("xxx");
//替换SecretKey为自己腾讯云账号下申请的密钥信息
context.setSecretKey("xxx");

//替换Region信息
context.setRegion("ap-shanghai");

 // 替换数据订阅连接信息，从详情页获取的ip，port
context.setServiceIp("172.17.16.2");
context.setServicePort(7507);
final DefaultSubscribeClient client = new DefaultSubscribeClient(context);

// 填写对应要同步的数据库和表名
//替换订阅的库名
final String targetDatabase = "tcaplus";
client.addClusterListener(listener);
 // 替换订阅的通道名，订阅详情页获取的 dts-channel 配置信息，填写到此处
client.askForGUID("dts-channel-5ljfLZmhFvnTCQx9");
client.start();

javac -cp binlogsdk-2.8.2-jar-with-dependencies.jar:kafka-clients-1.1.0.jar:slf4j-api-1.7.25.jar:slf4j-log4j12-1.7.2.jar:json-simple-1.1.jar  -encoding UTF-8  KafkaDemo.java

java -XX:-UseGCOverheadLimit -Xms2g -Xmx2g  -classpath .:binlogsdk-2.6.0-release.jar:kafka-clients-1.1.0.jar:slf4j-api-1.7.25.jar:slf4j-log4j12-1.7.2.jar:json-simple-1.1.jar  KafkaDemo

import json
import MySQLdb
#替换DB连接信息，从已申请的MySQL实例中获取
db = MySQLdb.connect("172.17.16.17", "xxx", "xxx", "tcaplus", charset='utf8' )

cursor = db.cursor()
#模拟一个json结构数据转成字符串后以byte形式写入MySQL
info={"player_phone":"123456","player_email":"June@123.com"}
info_bin = json.dumps(info).encode('utf-8')
sql = "INSERT INTO test (player_id, player_name, player_info) VALUES (%s, %s, %s)"
val = (7, 'June', info_bin,)

cursor.execute(sql, val)
db.commit()
db.close()

#替换MySQL连接账户名和密码
mysqldump -h172.17.16.17 -u[db_user] -p[db_password] -P3306 -B tcaplus --tables test  --skip-opt >test.sql

#替换用户名和密码，行之间间隔符}}}{{{
mysql -h172.17.16.17 -u[db_user] -p[db_password] -P3306 -D tcaplus -Ne "select JSON_OBJECT('player_id',player_id,'player_name',player_name,'player_info',player_info)  from test" >test.json

{"player_id": 1, "player_info": "base64:type15:eyJwbGF5ZXJfZW1haWwiOiAiMUB0ZXN0LmNvbSIsICJwbGF5ZXJfcGhvbmUiOiAiMTIzNDU2NiJ9", "player_name": "ball"}

import json
import base64
#引入TcaplusRestClient类
from tcaplusdb_client.tcaplusdb_rest_client import TcaplusRestClient
client = None
#替换TcaplusDB集群访问地址
endpoint = "http://172.17.16.15"
#替换TcaplusDB访问ID
access_id = 20
#替换TcaplusDB集群密码
access_password = "xxx"
#替换TcaplusDB表格组ID
tablegroup_id = 1
#替换TcaplusDB表名
table_name =  "test"
def init():
    #初始化TcaplusDB客户端
    global client
    client = TcaplusRestClient(endpoint, access_id, access_passwd)
    client.SetTargetTable(table_group_id=tablegroup_id, table_name=table_name)
def write():
    #从文件读数据并写入TcaplusDB
    filename='test.json'
    with open(filename,'r') as f:
        lines = f.readlines()
        for line in lines:
            json_str = json.loads(line)
            player_info = json_str['player_info'].replace("base64:type15:","")
            #将base64编码的字段反解码成字符串并转换成JSON格式
            json_str['player_info'] = json.loads(base64.b64decode(player_info))
            stat, resp = client.AddRecord(json_str)
if __name__ == '__main__':
    init()
    write()

pip install -U cos-python-sdk-v5

from qcloud_cos import CosConfig
from qcloud_cos import CosS3Client

#替换COS所在Region名
cosRegionName='ap-shanghai'
#替换腾讯云账户的密钥信息
SECRET_ID='xxx'
SECRET_KEY='xxx'

scheme='https'
config = CosConfig(Region=cosRegionName, SecretId=SECRET_ID, SecretKey=SECRET\_KEY,Scheme=scheme)
client = CosS3Client(config)

# 创建bucket, 注意格式为migrate名-appid (appid是自己腾讯云账号下的appid)
response = client.create_bucket(
    Bucket='migrate-1258272208'
)
#上传文件到COS
#替换bucket名，要上传的文件名及Key, PartSize指定分包大小(单位MB)，MAXThread指定并发上传的线程数
response = client.upload_file(
            Bucket='migrate-1258272208',
            LocalFilePath='1.txt',
            Key='1.txt',
            PartSize=1,
            MAXThread=10,
            EnableMD5=False
        )

#从COS下载文件
#替换Bucket名
response = client.get_object(
        Bucket='migrate-1258272208',
        Key='1.txt',
        )
response['Body'].get_stream_to_file('output.txt')