datax实现MySQL数据库迁移shell自动化脚本

datax实现MySQL数据库迁移

（1）生成python脚本

# coding=utf-8
import json
import getopt
import os
import sys
import MySQLdb#MySQL相关配置，需根据实际情况作出修改
mysql_host = "xxxx"
mysql_port = "3306"
mysql_user = "xxxx"
mysql_passwd = "xxxx"# MYSQL Destination
dest_mysql_host = "xxxx"
dest_mysql_port = "3306"
dest_mysql_user = "xxxx"
dest_mysql_passwd = "xxxx"#生成配置文件的目标路径，可根据实际情况作出修改
output_path = "/opt/module/datax/job/import/databaseName"def get_connection():return MySQLdb.connect(host=mysql_host, port=int(mysql_port), user=mysql_user, passwd=mysql_passwd)def get_mysql_meta(database, table):connection = get_connection()cursor = connection.cursor()sql = "SELECT COLUMN_NAME,DATA_TYPE from information_schema.COLUMNS WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s ORDER BY ORDINAL_POSITION"cursor.execute(sql, [database, table])fetchall = cursor.fetchall()cursor.close()connection.close()return fetchalldef get_mysql_columns(database, table):return map(lambda x: x[0], get_mysql_meta(database, table))def get_hive_columns(database, table):def type_mapping(mysql_type):mappings = {"bigint": "bigint","int": "bigint","smallint": "bigint","tinyint": "bigint","mediumint": "bigint","decimal": "string","double": "double","float": "float","binary": "string","char": "string","varchar": "string","datetime": "string","time": "string","timestamp": "string","date": "string","text": "string"}return mappings[mysql_type]meta = get_mysql_meta(database, table)return map(lambda x: {"name": x[0], "type": type_mapping(x[1].lower())}, meta)def generate_json(source_database, source_table):job = {"job": {"setting": {"speed": {"channel": 15},"errorLimit": {"record": 0,"percentage": 0.02}},"content": [{"reader": {"name": "mysqlreader","batchSize": "8192","batchByteSize": "33554432","parameter": {"username": mysql_user,"password": mysql_passwd,"column": get_mysql_columns(source_database, source_table),"splitPk": "","connection": [{"table": [source_table],"jdbcUrl": ["jdbc:mysql://" + mysql_host + ":" + mysql_port + "/" + source_database+"?userCompress=true&useCursorFetch=true&useUnicode=true&characterEncoding=utf-8&useSSL=false"]}]}},"writer": {"name": "mysqlwriter","batchSize": "8192","batchByteSize": "33554432","parameter": {"writeMode": "replace","username": dest_mysql_user,"password": dest_mysql_passwd,"column": get_mysql_columns(source_database, source_table),"connection": [{"jdbcUrl": "jdbc:mysql://"+dest_mysql_host+":"+dest_mysql_port+ "/" + source_database + "?userCompress=true&useCursorFetch=true&useUnicode=true&characterEncoding=utf-8&useSSL=false","table": [source_table],}]}},"transformer": [{"name": "dx_groovy","parameter": {"code": "for(int i=0;i<record.getColumnNumber();i++){if(record.getColumn(i).getByteSize()!=0){Column column = record.getColumn(i); def str = column.asString(); def newStr=null; newStr=str.replaceAll(\"[\\r\\n]\",\"\"); record.setColumn(i, new StringColumn(newStr)); };};return record;","extraPackage":[]}}]}]}}if not os.path.exists(output_path):os.makedirs(output_path)with open(os.path.join(output_path, ".".join([source_database, source_table, "json"])), "w") as f:json.dump(job, f)def main(args):source_database = ""source_table = ""options, arguments = getopt.getopt(args, '-d:-t:', ['sourcedb=', 'sourcetbl='])for opt_name, opt_value in options:if opt_name in ('-d', '--sourcedb'):source_database = opt_valueif opt_name in ('-t', '--sourcetbl'):source_table = opt_valuegenerate_json(source_database, source_table)if __name__ == '__main__':main(sys.argv[1:])

sh脚本

#!/bin/bash
python ~/bin/new_lms_mysql_gen_import_config.py -d database -t table

mysq到mysql脚本

#! /bin/bash
case $1 in
"course_groups")python /opt/module/datax/bin/datax.py /opt/module/datax/job/import/database/database.table.json
;;"all")python /opt/module/datax/bin/datax.py /opt/module/datax/job/import/database/database.table.json;;
esac