作者简介：刘亚辉，现就职于北京海天起点新疆办事处，OGG项目组成员。具有5年以上电信行业系统运维经验，熟悉Oracle数据库、Linux操作系统日常维护工作，擅长于OGG搭建方案的制定、运维优化及故障诊断。

最近在一个数据迁移的项目中碰到一个问题，项目中需要从MySQL数据库中同步一些表的数据到Oracle数据库，同步工具已经搞定，但是同步前需要先在Oracle创建表结构。由于MySQL跟Oracle之间在SQL语句方面有着一些差异（建表语句，字段类型等），需要做适当的转换。而且这种转换在今后会经常被用到（有新增的表同步需求时就会用到）。所以这种转换一定要通过程序/脚本自动化的方式完成，而且得满足以下几个条件：

• 健壮性
  

  转换程序/脚本在编写时需要尽可能地考虑到多种情况，尽量避免运行报错，或者转换出来的脚本无法执行的情况；
  

• 易用性
  

  不管编写什么程序/脚本，首要宗旨就是要好用。除了好用，操作步骤还不能太复杂，如果但凡使用一次就得先复制粘贴半天，那也就失去了意义。
  

最Easy的解决方案

基于这两点，首先想到的就是建表语句可以通过SQL拼接的方式完成。在Oracle数据库中，我们基本上都用过SQL拼接的方式来完成一些固定的工作。例如下面这个用来禁用外键的语句：

每次使用前只需要指定目标用户，就可以生成标准的外键禁用语句。在MySQL中，也有类似的视图。比如包含列相关信息的视图information_schema.columns：

图1 information_schema.columns视图字段

包含索引相关信息的视图information_schema.statistics：

图2 information_schema.statistics视图字段

最后再找到MySQL到Oracle字段的映射关系即可（MySQL -> Oracle）：

图3 常用的MySQL到Oracle的字段映射关系

基于以上分析，如果使用Python进行编写，只需要建立一个用于连接MySQL的实例类，获取建表所需的table name、column、index等信息，对数据类型进行转换（使用字典），再按照Oracle的格式进行拼接即可。

以上只提供了大概的实现方式，具体的实现代码就不做展示了。相信用这种最直观的方式实现的代码也是很好写的。下面主要说一下另一种”不那么简单”，却很有意思的方法（开始挖坑）。

关于表结构的获取方式

从事过Oracle数据库维护的人肯定都接触过dbms_metadata.get_ddl()这个函数，该函数可以通过SQL语句获取指定表或索引的建立语句。

可以发现，这个语句还包含一些关于段和存储方式相关的信息，这些信息在不显式指定的情况下都是按照默认值设置，而这里也一并输出了出来。如果我们能够批量得到这种标准的建表语句，再经过一定的数据处理，同样也是可以达到目的的。

mysqldump工具

所幸的是，MySQL也有相似的功能。不过不同于Oracle直接调用内置包中函数的方式，MySQL可以通过一种逻辑备份工具实现，即mysqldump。导出表结构的格式如下：

其中：

–no-data：表示不导出数据，只导出表结构；

-u：指定数据库登录用户；

-p：数据库用户对应的密码（如果当前用户是root，则无需指定）；

-P：指定端口；

db_name：表所在的数据库名称；

table_name：要导出的表名。

想查看其它参数及其详细用法，可以使用mysqldump – help命令。

在这里，我们只需要指定–no-data选项，以及要导出的表即可。我们可以先导出一张测试表，看看具体的格式：

图4 导出表结构

我去，这乱七八糟的什么鬼。。。如果第一次看到这些信息，肯定会有点犯迷糊，甚至会对此产生质疑：这显然没有第一种方法简便嘛，直接pass好了。

确实，论便捷性而言，这种方式没有第一种简单，还要额外处理数据格式的问题，但是却能体现出一种思维：数据分析的思维。

何为数据分析？

数据分析，随着近年来大数据的崛起，已经成为一门很火的专业。它的主要职能，简单来说，就是能够从海量不规则的原始数据中，提取出有价值的信息，并进行进一步的分析，体现出数据的潜在规律或者特性。数据分析的处理方式大致可以分为以下几个步骤：

• 获取原始数据
  

  通过各种方式获取未经处理的目标数据。了解过网络爬虫的小伙伴对这一步肯定非常了解，爬虫的第一步便是从互联网获取大量原始数据；
  

• 数据清洗
  

  虽然已获取了原始数据，但是其格式往往不符合我们的使用标准，需要剔除那些不需要的东西，并对格式做进一步的整理加工；
  

• 利用一些统计工具或者算法，对数据做进一步的分析，得到一些有用的结论。
  

在大概了解的数据分析的内容后，回到上面的疑问，我们何不将这个SQL语句转换的任务看做是一个数据分析的小项目呢？借此机会，可以训练一下我们处理数据的能力（实际上只需要完成前两步即可）。

总体步骤

按照以上数据分析的步骤，对应到我们这个项目中的步骤如下：

具体步骤

• • 使用mysqldump提取原始语句
    

    根据上面对mysqldump工具的简要介绍，我们只需要按照命令格式拼接好，并执行即可。但是有个问题，如何获取指定的MySQL数据库名和表名呢？有两种方式，一种是新建一个存放待提取数据库名和表名的配置文件，程序运行时直接读取该配置文件获取即可；还有一种是只指定数据库名，并建立数据库连接，获取下面的所有表名。我们将采用后一种方式，顺便熟悉一下建立数据库连接的方式。
    

    1. 创建数据库连接类：
       
       import pymysql # Python用于连接MySQL的包 class ConnInst: def __init__(self, connect_dict): # 根据数据库登录所需信息(均存放在connect_dict字典中)，创建用于连接的conn类属性 self.conn = pymysql.connect( host=connect_dict['host'], port=connect_dict['port'], user=connect_dict['username'], password=connect_dict['password'], db=connect_dict['db'], charset=connect_dict['charset'] ) # 用于执行SQL语句的数据库cursor self.cursor = self.conn.cursor() def execute_query(self, sql_string): # 创建用于执行查询语句的方法 try: cursor = self.cursor cursor.execute(sql_string) result_list = cursor.fetchall() cursor.close() self.conn.close() return result_list except pymysql.Error as e: print("mysql execute error: ", e) raise def execute_dml(self, sql_string): # 创建用于执行DML操作的方法（区别在于有提交操作，本例中不会使用此方法） try: cursor = self.cursor cursor.execute(sql_string) self.conn.commit() cursor.close() self.conn.close() except pymysql.Error as e: print("mysql execute error: ", e) raise

       1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41

       import pymysql  # Python用于连接MySQL的包       class ConnInst:     def __init__(self, connect_dict):         # 根据数据库登录所需信息(均存放在connect_dict字典中)，创建用于连接的conn类属性         self.conn = pymysql.connect(             host=connect_dict['host'],             port=connect_dict['port'],             user=connect_dict['username'],             password=connect_dict['password'],             db=connect_dict['db'],             charset=connect_dict['charset']         )         # 用于执行SQL语句的数据库cursor         self.cursor = self.conn.cursor()        def execute_query(self, sql_string):         # 创建用于执行查询语句的方法         try:             cursor = self.cursor             cursor.execute(sql_string)             result_list = cursor.fetchall()             cursor.close()             self.conn.close()             return result_list         except pymysql.Error as e:             print("mysql execute  error: ", e)             raise        def execute_dml(self, sql_string):         # 创建用于执行DML操作的方法（区别在于有提交操作，本例中不会使用此方法）         try:             cursor = self.cursor             cursor.execute(sql_string)             self.conn.commit()             cursor.close()             self.conn.close()         except pymysql.Error as e:             print("mysql execute  error: ", e)             raise

    2. 创建获取未经处理的MySQL建表SQL语句的方法：
       
       import os def get_sql_unformatted(conn, conn_dict, mysql_file_path, mysql_statement): # 方法中定义的形参： # conn: 数据库cursor # conn_dict： 存放数据库连接信息的字典 # mysql_file_path: 指定mysqldump命令生成的批量语句文件的存放路径 # mysql_statement: 批量生成的建表语句文件 # 拼接根据数据库名获取所有表名的语句 sql_get_tables = "select table_name from information_schema.`TABLES` " \ "where table_schema = '" + conn_dict['db'] + "';" # 将获取的表名存放在result_list列表中 result_list = conn.execute_query(sql_get_tables) # 判断存放路径是否存在，如果不存在则新建一个 if not os.path.exists(mysql_file_path): os.mkdir(mysql_file_path) # 存放mysqldump导出信息的字典 dump_cmd_dict = { 'dumpcmd': 'mysqldump --no-data', 'server': conn_dict['host'], 'user': conn_dict['username'], 'password': conn_dict['password'], 'port': conn_dict['port'], 'db': conn_dict['db'] } sqldump_cmds = '' if result_list: os.chdir(mysql_file_path) for row in result_list: # 获取表名 table_name = row[0] # mysqldump命令格式 sqlformat = '%s -h%s -u%s -p%s -P%s %s %s >> %s' # 生成相应的sql语句 sql = [sqlformat % ( dump_cmd_dict['dumpcmd'], dump_cmd_dict['server'], dump_cmd_dict['user'], dump_cmd_dict['password'], dump_cmd_dict['port'], dump_cmd_dict['db'], table_name, mysql_statement )] tmp = sql[0] + '\n' # 每行末尾增加换行符 sqldump_cmds += tmp # 循环执行导出语句，并将结果输出到mysql_statement文件中 for sqldump_cmd in sqldump_cmds: os.system(sqldump_cmd)

       1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52

       import os    def get_sql_unformatted(conn, conn_dict, mysql_file_path, mysql_statement):     # 方法中定义的形参：         # conn: 数据库cursor         # conn_dict： 存放数据库连接信息的字典         # mysql_file_path: 指定mysqldump命令生成的批量语句文件的存放路径         # mysql_statement: 批量生成的建表语句文件        # 拼接根据数据库名获取所有表名的语句     sql_get_tables = "select table_name from information_schema.`TABLES` " \                      "where table_schema = '" + conn_dict['db'] + "';"     # 将获取的表名存放在result_list列表中     result_list = conn.execute_query(sql_get_tables)     # 判断存放路径是否存在，如果不存在则新建一个     if not os.path.exists(mysql_file_path):         os.mkdir(mysql_file_path)        # 存放mysqldump导出信息的字典     dump_cmd_dict = {         'dumpcmd': 'mysqldump --no-data',         'server': conn_dict['host'],         'user': conn_dict['username'],         'password': conn_dict['password'],         'port': conn_dict['port'],         'db': conn_dict['db']     }        sqldump_cmds = ''     if result_list:         os.chdir(mysql_file_path)         for row in result_list:             # 获取表名             table_name = row[0]             # mysqldump命令格式             sqlformat = '%s -h%s -u%s -p%s -P%s %s %s >> %s'             # 生成相应的sql语句             sql = [sqlformat % (                 dump_cmd_dict['dumpcmd'],                 dump_cmd_dict['server'],                 dump_cmd_dict['user'],                 dump_cmd_dict['password'],                 dump_cmd_dict['port'],                 dump_cmd_dict['db'],                 table_name,                 mysql_statement             )]             tmp = sql[0] + '\n' # 每行末尾增加换行符             sqldump_cmds += tmp         # 循环执行导出语句，并将结果输出到mysql_statement文件中         for sqldump_cmd in sqldump_cmds:             os.system(sqldump_cmd)

    3. 实例化类并调用方法，生成文件：
       
       def main(): # 设置MySQL连接信息： conn_dict = { 'host': '10.1.1.10', 'port': 8878, 'username': 'goldengate', 'password': 'goldengate123', 'db': 'order', 'charset': 'utf8' } # 设置结果存放路径及文件名 mysql_file_path = 'result_files' mysql_statement = 'mysql_ddl.sql' conn = ConnInst(conn_dict) # 生成Mysql导出表结构的命令至mysql_file_path的mysql_cmd_file文件 get_sql_unformatted(conn, conn_dict, mysql_file_path, mysql_statement) if __name__ == '__main__': main()

       1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21

       def main():     # 设置MySQL连接信息：     conn_dict = {         'host': '10.1.1.10',         'port': 8878,         'username': 'goldengate',         'password': 'goldengate123',         'db': 'order',         'charset': 'utf8'     }     # 设置结果存放路径及文件名     mysql_file_path = 'result_files'     mysql_statement = 'mysql_ddl.sql'        conn = ConnInst(conn_dict)     # 生成Mysql导出表结构的命令至mysql_file_path的mysql_cmd_file文件     get_sql_unformatted(conn, conn_dict, mysql_file_path, mysql_statement)       if __name__ == '__main__':     main()

       
       打开得到的文件后，格式就跟我们第一次看到的那样。接下来就需要对这些数据做数据清洗操作了。
       
  • 对原始语句按照规则进行数据清洗
    
    数据清洗常用的方法是利用正则表达式进行匹配，并进行清理、替换等操作。我们可以先分析一下原始数据的格式，看看哪些是需要清理的。
    
    1. 1. 注释类内容
          

          开头的这些注释掉的内容肯定是不需要的，其中分两种注释格式：单行注释”–“和多行注释”/* */”，所以以”–“和”/”开头的内容都是需要清除的；
          

       2. SET 语句
          
          
          SET语句的作用是设置各种变量，这个在Oracle中是不包含的，所以也可以剔除。
          在浏览数据时，还存在一个问题，就是SET语句如果过长，会做换行处理，这样一来，换行后的第二行就匹配不到了：
          
          但是通过观察可以发现，这些数据都是以’;结尾的，而正常的建表语句是不包含的，所以可以另写正则进行匹配；
          
       3. DROP TABLE语句
          

          MySQL想得很周到，在创建表前会先drop掉已存在的表，但我们的目的就是要新建表，所以就不用再drop了；
          

       4. Table name两边的`字符
          

          Oracle建表语句中不包含此类字符，所以可以一并替换掉；
          

       5. 替换DEFAULT CURRENT_TIMESTAMP
          

          MySQL取当前时间的关键字是CURRENT_TIMESTAMP ，Oracle中对应的是SYSDATE；而且Oracle中也不支持NOT NULL 与 DEFAULT同时使用，所以也要做替换；
          

       6. 替换MySQL指定存储引擎的相关语句
          

          在我们提取出来的数据中，每个建表语句的末尾都会指定存储引擎，Oracle中没有存储引擎的概念，所以可以直接替换掉；
          

          
          

       7. 删除空行
          

          经过上面一大堆的替换，肯定产生了大量的空行，为了保持语句的整洁，我们最后再统一处理。
          

    根据以上的分析，只需处理这7类数据即可：

    import re def data_clean(unformatted_sql_file, formatted_sql_file): with open(unformatted_sql_file, 'r', encoding='utf-8') as file_obj: lines = file_obj.readlines() sql_text = '' for line in lines: # 将注释、SET语句、DROP语句等字符用正则进行匹配，并替换为空 pattern1 = re.compile('--.*|/\*.*|SET @.*|DROP TABLE.*|^.*?\';$|`|', re.S) tmp = re.sub(pattern1, '', line) # 替换ENGINE所在行 pattern2 = re.compile('^\) ENGINE.*?;$', re.S) tmp = re.sub(pattern2, ');', tmp) # 替换 NOT NULL DEFAULT tmp = tmp.replace('NOT NULL DEFAULT', 'DEFAULT') # 替换当前时间 CURRENT_TIMESTAMP tmp = tmp.replace('CURRENT_TIMESTAMP', 'SYSDATE') # 如果是空行，则删除 if tmp == '\n': tmp = tmp.strip("\n") sql_text += tmp # 将结果保存到指定文件 with open(formatted_sql_file, 'w', encoding='utf-8') as file_obj: file_obj.write(sql_text)

    1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28

    import re       def data_clean(unformatted_sql_file, formatted_sql_file):     with open(unformatted_sql_file, 'r', encoding='utf-8') as file_obj:         lines = file_obj.readlines()        sql_text = ''     for line in lines:         # 将注释、SET语句、DROP语句等字符用正则进行匹配，并替换为空         pattern1 = re.compile('--.*|/\*.*|SET @.*|DROP TABLE.*|^.*?\';$|`|', re.S)         tmp = re.sub(pattern1, '', line)         # 替换ENGINE所在行         pattern2 = re.compile('^\) ENGINE.*?;$', re.S)         tmp = re.sub(pattern2, ');', tmp)         # 替换 NOT NULL DEFAULT         tmp = tmp.replace('NOT NULL DEFAULT', 'DEFAULT')         # 替换当前时间 CURRENT_TIMESTAMP         tmp = tmp.replace('CURRENT_TIMESTAMP', 'SYSDATE')         # 如果是空行，则删除         if tmp == '\n':             tmp = tmp.strip("\n")            sql_text += tmp        # 将结果保存到指定文件     with open(formatted_sql_file, 'w', encoding='utf-8') as file_obj:         file_obj.write(sql_text)

    该方法的第一个形参是输入的原始数据文件，第二个是输出的经处理后的文件。
    

    
    

    其实还有一个关于索引的问题，原始数据中包含建立索引的语句（KEY开头的语句），而Oracle不支持在建表语句中同时建索引，所以这部分要单独提取出来，生成建索引的语句。由于篇幅原因，这里就不详细说明了，感兴趣的小伙伴可以自己尝试实现一下。
    

    经过以上的处理，我们能得到的语句如下：
    

    只剩下字段数据类型的转换了。
    

  • MySQL到Oracle的字段转换
  • 关于字段的转换，这里给出一个处理思路。可以先定义一个只用于转换字段的函数，然后利用正则表达式，匹配出每张表中字段对应的行，再针对每一行的数据类型部分，循环调用转换函数即可完成替换。这里只给出字段转换函数的代码：
    def datatype_transfer(col_MySQL): # 定义MySQL到Oracle常用字段的映射关系 trans_dict = { 'BIGINT': 'NUMBER', 'BIT': 'RAW', 'BLOB': 'BLOB', 'DATETIME': 'DATE', 'DECIMAL': 'NUMBER', 'DOUBLE': 'NUMBER', 'DOUBLE PRECISION': 'NUMBER', 'ENUM': 'VARCHAR2', 'INT': 'NUMBER', 'INTEGER': 'NUMBER', 'LONGBLOB': 'BLOB', 'LONGTEXT': 'CLOB', 'MEDIUMBLOB': 'BLOB', 'MEDIUMINT': 'NUMBER', 'MEDIUMTEXT': 'CLOB', 'NUMERIC': 'NUMBER', 'REAL': 'FLOAT(24)', 'SET': 'VARCHAR2', 'SMALLINT': 'NUMBER', 'TEXT': 'VARCHAR2(4000)', 'TIME': 'DATE', 'TIMESTAMP': 'DATE', 'TINYBLOB': 'RAW', 'TINYINT': 'NUMBER', 'TINYTEXT': 'VARCHAR2', 'VARCHAR': 'VARCHAR2', 'YEAR': 'NUMBER', } if '(' in col_MySQL: # 如果是带括号的，说明数据类型是带长度的(如bigint(12))；否则是像时间等数据类型（如date），需要分别处理 dt_list = col_MySQL.split('(') if dt_list[0] not in trans_dict.keys(): # 如果字典的key中不包含dt_list[0]，说明缺少对应的字段转换关系，需要补录 print(dt_list[0] + ' is not in the transfer dict yet, please ' 'check.') # 将数据类型和长度拼接后再返回 return dt_list[0] + '(' + dt_list[1] else: return trans_dict[dt_list[0]] + '(' + dt_list[1] else: if col_MySQL not in trans_dict.keys(): print(col_MySQL + ' is not in the transfer dict yet, please check.') return col_MySQL else: return trans_dict[col_MySQL]

    1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30

    def datatype_transfer(col_MySQL):     # 定义MySQL到Oracle常用字段的映射关系     trans_dict = {         'BIGINT': 'NUMBER', 'BIT': 'RAW', 'BLOB': 'BLOB', 'DATETIME': 'DATE',         'DECIMAL': 'NUMBER', 'DOUBLE': 'NUMBER', 'DOUBLE PRECISION': 'NUMBER',         'ENUM': 'VARCHAR2', 'INT': 'NUMBER', 'INTEGER': 'NUMBER',         'LONGBLOB': 'BLOB', 'LONGTEXT': 'CLOB', 'MEDIUMBLOB': 'BLOB',         'MEDIUMINT': 'NUMBER', 'MEDIUMTEXT': 'CLOB', 'NUMERIC': 'NUMBER',         'REAL': 'FLOAT(24)', 'SET': 'VARCHAR2', 'SMALLINT': 'NUMBER',         'TEXT': 'VARCHAR2(4000)', 'TIME': 'DATE', 'TIMESTAMP': 'DATE',         'TINYBLOB': 'RAW', 'TINYINT': 'NUMBER', 'TINYTEXT': 'VARCHAR2',         'VARCHAR': 'VARCHAR2', 'YEAR': 'NUMBER',     }     if '(' in col_MySQL:         # 如果是带括号的，说明数据类型是带长度的(如bigint(12))；否则是像时间等数据类型（如date），需要分别处理         dt_list = col_MySQL.split('(')         if dt_list[0] not in trans_dict.keys():             # 如果字典的key中不包含dt_list[0]，说明缺少对应的字段转换关系，需要补录             print(dt_list[0] + ' is not in the transfer dict yet, please '                                'check.')             # 将数据类型和长度拼接后再返回             return dt_list[0] + '(' + dt_list[1]         else:             return trans_dict[dt_list[0]] + '(' + dt_list[1]     else:         if col_MySQL not in trans_dict.keys():             print(col_MySQL + ' is not in the transfer dict yet, please check.')             return col_MySQL         else:             return trans_dict[col_MySQL]

• 1. 这个函数的形参是MySQL字段类型，返回的是Oracle数据类型。最后再经过一轮循环替换，就可以得到Oracle的建表语句：
     美中不足的是没有明确指定用户名和表空间，只能放在特定的用户下取执行。对代码稍作修改也可实现。
     

总结

这篇文章看似是解决数据库字段转换问题的，实际上是来做数据分析入门的。急需解决方法的，看第一种即可，后面均可忽略。后面这种方法是抛弃了使用数据库内置视图的捷径，而完全采用数据分析的处理思路完成的，而且涵盖了常用的数据处理方法。虽然繁冗不堪，但值得对此有兴趣的同仁一看。增加对数据的敏感度，时间长了，也许就能看见一个不一样的世界。

未经允许不得转载：Oracle一体机用户组 » MySQL to Oracle 字段转换