poyee-data-warehouse/bin/datax-gc-generator.py

#!/usr/bin/env /usr/bin/python3
# -*- coding:utf-8 -*-
"""
  生成DataX作业配置文件生成器的配置文件
"""
# -*- coding=utf-8 -*-

import os
import re
import sys
import time

base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(base_dir)
from dw_base import DO_RESET, NORM_CYN, NORM_GRN, NORM_MGT, NORM_YEL, NORM_RED
from dw_base.database.mysql_utils import MySQLHandler
from dw_base.datax.datasources.hdfs_data_source import HDFSDataSource
from dw_base.datax.datasources.mysql_data_source import MySQLDataSource
from dw_base.datax.datax_utils import convert_mysql_column_types
from dw_base.datax.plugins.reader.hdfs_reader import HDFSReader
from dw_base.datax.plugins.reader.mysql_reader import MySQLReader
from dw_base.datax.plugins.writer.hbase_writer import HBaseWriter
from dw_base.datax.plugins.writer.hdfs_writer import HDFSWriter
from dw_base.datax.plugins.writer.kafka_writer import KafkaWriter
from dw_base.datax.plugins.writer.mongo_writer import MongoWriter, MONGO_SPECIAL_WORDS_DICT
from dw_base.hive.hive_utils import get_hive_database_name, get_hive_table_prefix
from dw_base.spark.spark_sql import SparkSQL
from dw_base.utils.common_utils import exist
from dw_base.utils.config_utils import parse_args
from dw_base.utils.file_utils import write_file, append_file, list_files, load_json_file
from dw_base.utils.log_utils import pretty_print
from dw_base.utils.string_utils import snake_case_to_pascal_case, snake_case_to_camel_case


def usage(code: int):
    print(
        f'{NORM_MGT}Usage: {sys.argv[0]}\n'
        f'{NORM_CYN}\t[-H/--H/--help]                   打印脚本使用方法{DO_RESET}'
    )
    if not from_system or not to_system:
        print(
            f'{NORM_MGT}Usage: {sys.argv[0]}\n'
            f'{NORM_GRN}\t<[-]-from< /=>from>               源系统类型（默认MySQL），目前支持hdfs、mysql\n'
            f'{NORM_GRN}\t<[-]-to< /=>to>                   目标系统类型（默认HDFS），目前支持elasticsearch、hbase、hdfs、kafka、mongo、mysql\n'
            f'{NORM_CYN}\t[[-]-output< /=>output directory] 生成的ini文件存储位置，可以是绝对路径或相对路径'
            f'{DO_RESET}'
        )
    if from_system == "hdfs":
        print(
            f'{NORM_MGT}Parameters when from is HDFS: \n'
            f'{NORM_GRN}\t<[-]-d< /=>database>              Hive数据库（默认crl_mg）\n'
            f'{NORM_GRN}\t<[-]-t< /=>table>                 Hive数据表（可多次传入，不传扫描-D传入的库中所有表）\n'
            f'{NORM_GRN}\t<[-]-e< /=>exclude>               忽略的Hive数据表（可多次传入）\n'
            f'{NORM_CYN}\t[[-]-partitioned]                 是否是分区表（默认为是），目前只支持日分区\n'
            f'{DO_RESET}'
        )
    elif from_system == "mysql":
        print(
            f'{NORM_MGT}Parameters when from is MySQL: \n'
            f'{NORM_GRN}\t<[-]-h< /=>host>                  MySQL主机\n'
            f'{NORM_CYN}\t[[-]-P< /=>port]                  MySQL端口\n'
            f'{NORM_GRN}\t<[-]-u< /=>username>              MySQL用户\n'
            f'{NORM_GRN}\t<[-]-p< /=>password>              MySQL密码\n'
            f'{NORM_GRN}\t<[-]-D< /=>database>              MySQL数据库\n'
            f'{NORM_GRN}\t<[-]-tr< /=>table>                MySQL数据表正则（可多次传入，优先级高于t）\n'
            f'{NORM_GRN}\t<[-]-t< /=>table>                 MySQL数据表（可多次传入）\n'
            f'{NORM_GRN}\t<[-]-er< /=>exclude>              忽略的MySQL数据表正则（可多次传入，优先级高于e）\n'
            f'{NORM_GRN}\t<[-]-e< /=>exclude>               忽略的MySQL数据表（可多次传入）\n'
            f'{NORM_CYN}\t[[-]-inc-col]                     增量抽取字段名称（默认update_time）'
            f'{DO_RESET}'
        )
    if to_system == "elasticsearch":
        print(
            f'{NORM_MGT}Parameters when to is Elasticsearch: \n'
            f'{NORM_CYN}\t[[-]-inc-col]                     增量抽取字段名称（默认update_time）'
            f'{DO_RESET}'
        )
    elif to_system == "hbase":
        print(
            f'{NORM_MGT}Parameters when to is HBase: \n'
            f'{NORM_CYN}\t[[-]-inc-col]                     增量抽取字段名称（默认update_time）'
            f'{DO_RESET}'
        )
    elif to_system == "hdfs":
        print(
            f'{NORM_MGT}Parameters when to is HDFS: \n'
            f'{NORM_CYN}\t[[-]-project< /=>project]         项目名称，如skb、bms\n'
            f'{NORM_CYN}\t[[-]-layer< /=>dw-layer]          数据仓库分层，如ods、dwd\n'
            f'{NORM_CYN}\t[[-]-env< /=>dw-env]              数据仓库分层，如test\n'
            f'{NORM_CYN}\t[[-]-edition< /=>dw-edition]      数据仓库版本，如dl,dd,\n'
            f'{NORM_CYN}\t[[-]-partitioned]                 是否是分区表（默认为是），目前只支持日分区'
            f'{DO_RESET}'
        )
    elif to_system == "kafka":
        print(
            f'{NORM_MGT}Parameters when to is Kafka: \n'
            f'{NORM_GRN}\t<[-]-T< /=>kafka topic>           Kafka Topic\n'
            f'{NORM_GRN}\t<[-]-K< /=>kafka key>             Kafka Key of data\n'
            f'{DO_RESET}'
        )
    elif to_system == "mongo":
        print(
            f'{NORM_MGT}Parameters when to is Mongo: \n'
            f'{NORM_CYN}\t[[-]-inc-col]                     增量抽取字段名称（默认update_time）'
            f'{DO_RESET}'
        )
    elif to_system == "mysql":
        print(
            f'{NORM_MGT}Parameters when to is MySQL: \n'
            f'{NORM_GRN}\t<[-]-h< /=>host>                  MySQL主机\n'
            f'{NORM_CYN}\t[[-]-P< /=>port]                  MySQL端口\n'
            f'{NORM_GRN}\t<[-]-u< /=>username>              MySQL用户\n'
            f'{NORM_GRN}\t<[-]-p< /=>password>              MySQL密码\n'
            f'{NORM_GRN}\t<[-]-D< /=>database>              MySQL数据库\n'
            f'{NORM_GRN}\t<[-]-t< /=>table>                 MySQL数据表'
            f'{DO_RESET}'
        )
    exit(code)


def hdfs_elasticsearch_generator():
    raise Exception(f'Not implemented yet with from {from_system} and to {to_system}')


def hdfs_hbase_generator():
    with SparkSQL('datax-gc-generator') as spark_sql:
        hdfs_ds_name = 'hdfs-aliyun-cloud'
        hbase_ds_name = 'hbase-default'
        hdfs_path = '/user/hive/warehouse'
        output = CONFIG.get('output', f'{base_dir}/ignored')
        hive_database = CONFIG.get('d')
        if not hive_database:
            pretty_print(f'{NORM_RED}参数 {NORM_GRN}-d{DO_RESET}{NORM_RED} 未提供')
            usage(1)
        config_ini_path = '{0}/config/hdfs-hbase/{1}'.format(output, hive_database)
        os.system(f'mkdir -p {config_ini_path}')
        included_tables = CONFIG.get('t', [])
        if isinstance(included_tables, str):
            included_tables = [included_tables]
        excluded_tables = CONFIG.get('e', [])
        if isinstance(excluded_tables, str):
            excluded_tables = [excluded_tables]
        if len(included_tables) == 0 and len(excluded_tables) == 0:
            pretty_print(f'{NORM_YEL}注意：'
                         f'{NORM_MGT}参数 {NORM_GRN}-t{NORM_MGT} 或 {NORM_GRN}-e '
                         f'{NORM_MGT}未提供，将扫描数据库 '
                         f'{NORM_GRN}{hive_database}{DO_RESET} '
                         f'{NORM_MGT}下所有表')
        hbase_namespace = CONFIG.get('n')
        if not hbase_namespace:
            pretty_print(f'{NORM_YEL}注意：'
                         f'{NORM_MGT}参数 {NORM_GRN}-n{DO_RESET} '
                         f'{NORM_MGT}未提供，将使用 HBase 默认命名空间 '
                         f'{NORM_GRN}default')
            hbase_namespace = 'default'
        final_tables = []
        if len(included_tables) == 0:
            all_hive_tables = []
            # 如果没有传入仅需的表则获取数据库下所有表
            tables_df, _ = spark_sql.query(f'show tables in {hive_database}', silent=True)
            desc_tables = tables_df.collect()
            for row in desc_tables:
                all_hive_tables.append(row[1])
            # 去除忽略的数据表
            for hive_table in all_hive_tables:
                if hive_table not in excluded_tables:
                    final_tables.append(f'{hive_database}.{hive_table}')
        else:
            final_tables = included_tables
        number_width = len(str(len(final_tables)))
        index = 1
        for hive_table_full_name in final_tables:
            if hive_table_full_name.__contains__("."):
                hive_db, hive_table_name = hive_table_full_name.split('.')
            else:
                hive_db = hive_database
                hive_table_name = hive_table_full_name
            # 获取表备注
            create_table_str = spark_sql.query_scalar(f'show create table {hive_table_full_name}', silent=True)
            hive_table_comment = create_table_str.split('ROW FORMAT SERDE')[0].split(")")[1]
            if 'COMMENT' in hive_table_comment:
                hive_table_comment = hive_table_comment.split("'")[1]
            else:
                hive_table_comment = ''
            # 表是否有分区
            partitioned = False
            if "`dt`" in create_table_str:
                partitioned = True
            # 获取表列名
            hive_columns = spark_sql.get_columns(f'{hive_table_full_name}')
            hive_column_names = []
            hive_column_types = {}
            for column_name, column_value in hive_columns.items():
                hive_column_names.append(column_name)
                column_type = column_value[0].upper()
                if column_type == 'BIGINT':
                    column_type = 'LONG'
                hive_column_types[column_name] = column_type
            hdfs_reader_str = HDFSReader.generate_definition(hdfs_ds_name, hdfs_path, hive_db, hive_table_name,
                                                             hive_table_comment,
                                                             partitioned,
                                                             hive_column_names, hive_column_types)
            # hbase列名
            hbase_column_names = []
            for column_name in hive_column_names:
                hbase_column_names.append(f'cf:{column_name}')
            row_key_columns = []
            row_key_columns.append('reverse(主键如果是自增ID，建议reverse)')
            row_key_columns.append('separator(@@)')
            row_key_columns.append(f'separator({hive_db})')
            row_key_columns.append(f'separator(.)')
            row_key_columns.append(f'separator({hive_table_name})')
            hbase_writer_str = HBaseWriter.generate_definition(hbase_ds_name,
                                                               hbase_namespace,
                                                               hive_table_name,
                                                               hive_table_name,
                                                               hive_table_comment,
                                                               "cf",
                                                               hbase_column_names,
                                                               {},
                                                               row_key_columns)
            # 写文件
            generator_config_file = '{0}/hdfs-hbase-{1}-{2}.ini'.format(config_ini_path, hbase_namespace, hive_table_name)
            write_file(f'{hdfs_reader_str}\n\n{hbase_writer_str}\n', generator_config_file)
            pretty_print(
                f'{NORM_YEL}{str(index).rjust(number_width, " ")}. {NORM_MGT} 读 Hive 数据库表 '
                f'{NORM_GRN}{hive_database}.{hive_table_name}  '
                f'{NORM_MGT}\n                    写 HBase DataX作业配置文件生成器配置已写入文件 '
                f'{NORM_GRN}{generator_config_file}'
            )
            index += 1


def hdfs_kafka_generator():
    kafka_topic = CONFIG.get('T')
    kafka_key = CONFIG.get('K')
    if not kafka_topic:
        pretty_print(f'{NORM_RED}参数 {NORM_GRN}-T{DO_RESET}{NORM_RED} 未提供')
        usage(1)
    included_tables = CONFIG.get('t', [])
    if isinstance(included_tables, str):
        included_tables = [included_tables]
    excluded_tables = CONFIG.get('e', [])
    if isinstance(excluded_tables, str):
        excluded_tables = [excluded_tables]
    hive_database = CONFIG.get('d')
    if len(included_tables) == 0 and len(excluded_tables) == 0:
        if not hive_database:
            pretty_print(f'{NORM_RED}参数 {NORM_GRN}-d{DO_RESET}{NORM_RED} 未提供')
            usage(1)
        else:
            pretty_print(f'{NORM_YEL}注意：'
                         f'{NORM_MGT}参数 {NORM_GRN}-t{NORM_MGT} 或 {NORM_GRN}-e '
                         f'{NORM_MGT}未提供，将扫描数据库 '
                         f'{NORM_GRN}{hive_database}{DO_RESET} '
                         f'{NORM_MGT}下所有表')
    with SparkSQL('datax-gc-generator') as spark_sql:
        hdfs_ds_name = 'hdfs-aliyun-cloud'
        hdfs_path = '/user/hive/warehouse'
        kafka_ds_name = 'kafka-aliyun'
        output = CONFIG.get('output', f'{base_dir}/ignored')
        final_tables = []
        if len(included_tables) == 0:
            all_hive_tables = []
            # 如果没有传入仅需的表则获取数据库下所有表
            tables_df, _ = spark_sql.query(f'show tables in {hive_database}', silent=True)
            desc_tables = tables_df.collect()
            for row in desc_tables:
                all_hive_tables.append(row[1])
            # 去除忽略的数据表
            for hive_table in all_hive_tables:
                if hive_table not in excluded_tables:
                    final_tables.append(f'{hive_database}.{hive_table}')
        else:
            final_tables = included_tables
        number_width = len(str(len(final_tables)))
        index = 1
        # 判断是否为外部表，若为外部表，则获取表信息内容改变
        for table_name in final_tables:
            # if hive_table_full_name.endswith("es_mapping"):
            #     continue
            # hive_database, hive_table_name = hive_table_full_name.split('.')  # type: str,str
            config_ini_path = '{0}/config/hdfs-kafka/{1}'.format(output, hive_database)
            os.system(f'mkdir -p {config_ini_path}')
            hive_table_name = ''
            # 如果传入的配置中（-t=*）是完整的库名+表名（crl_es.xxx)，则通过完整的库名表名获取信息
            if table_name.__contains__('.'):
                hive_database, hive_table_name = table_name.split('.')  # type: str,str
                hive_table_full_name = table_name
            # 如果传入的配置中是表名，则通过传入的配置获取库信息
            elif hive_database is not None:
                hive_table_name = table_name  # type: str
                hive_table_full_name = f'{hive_database}.{hive_table_name}'
            else:
                raise ValueError("hive_database undefined")
            hive_column_names = []
            hive_column_types = {}
            hive_table_names_mapping = {}
            # 获取表备注
            if hive_table_full_name.__contains__('mapping'):
                mysql_handler = MySQLHandler(
                    host='rm-m5er2i6wz605su9bi.mysql.rds.aliyuncs.com',
                    port=3306,
                    username='meta_ro',
                    password='Ts#r5rO1'
                )
                partitioned = True  # 如果是mapping表则默认有分区，因为只是通过mapping表看映射
                hive_table_tbl = dict(
                    mysql_handler.query_tbl_hive_metadata(hive_table_name))  # 获取hive的mapping表全部的tblproperties
                hive_table_comment = hive_table_tbl.get('comment')  # 获取hive表comment字段
                hive_table_names_mapping = hive_table_tbl.get('es.mapping.names')  # 获取hive与es映射的字段信息
                hive_table_es_index = hive_table_tbl.get('es.resource').split('/')[0]  # 获取写入es的索引
                hive_table_column = mysql_handler.query_column_hive_metadata(hive_table_name)  # 获取hive表的列名、字段类型、注释等数据
                hive_columns = {column[1]: (column[2], column[3]) for column in
                                hive_table_column}  # 将读取到的列名、类型、注释转换成与spark读取出来的格式一致
                kafka_column_types = {column[1]: column[2] for column in hive_table_column}  # 单独列出一个{列名：类型}的字典
                hdfs_reader_column = [mappingName.split(':')[0] for mappingName in
                                      str(hive_table_names_mapping).split(',')]  # 在hive与es映射字段的表内取出要读hdfs的字段
                kafka_writer_column = [mappingName.split(':')[1] for mappingName in
                                       str(hive_table_names_mapping).split(',')]  # 在hive与es映射字段的表内取出要写kafka的字段
                column_mapping = dict(item.split(":") for item in hive_table_names_mapping.split(","))  # 将hive与es字段的映射转换为字典

                for k, v in dict(
                        item.split(":") for item in hive_table_names_mapping.split(",")).items():  # 保存一个不包含struct结构体映射的字典
                    if k not in hive_columns.keys():
                        column_mapping.pop(k)

                for column in hdfs_reader_column.copy():  # HDFS reader中去掉结构体的列名
                    if column not in hive_columns.keys():
                        hdfs_reader_column.remove(column)

                for column in kafka_writer_column.copy():  # kafka reader中去掉结构体的列名
                    if column not in column_mapping.values():
                        kafka_writer_column.remove(column)

                hive_es_column_mapping = {hdfs_reader_column[i]: kafka_writer_column[i] for i in
                                          range(len(hdfs_reader_column))}

                for key, value in kafka_column_types.copy().items():  # 替换kafka writer字典中的键
                    if key in hive_es_column_mapping:
                        kafka_column_types[hive_es_column_mapping[key]] = kafka_column_types.pop(key)
                    if key not in hive_es_column_mapping:
                        kafka_column_types.pop(key)

                hive_column_types = {}
                column_type_flag = ['LONG', 'BIGINT', 'BOOLEAN', 'STRING', 'DOULBE']
                for column_name, column_value in hive_columns.items():
                    hive_column_names.append(column_name)
                    column_type = column_value[0].upper()
                    if column_type in column_type_flag:
                        hive_column_types[column_name] = column_type
                    else:
                        if column_type == 'INT':
                            hive_column_types[column_name] = 'LONG'

                hive_column_names = hdfs_reader_column
                hive_table_name = re.sub(r"({}).*$".format(re.escape("sum")), "sum", hive_table_name)
                if kafka_key is None:
                    kafka_key = hive_table_es_index
            else:
                create_table_str = spark_sql.query_scalar(f'show create table {hive_table_full_name}', silent=True)
                hive_table_comment = create_table_str.split('ROW FORMAT SERDE')[0].split(")")[1]
                if 'COMMENT' in hive_table_comment:
                    hive_table_comment = hive_table_comment.split("'")[1]
                else:
                    hive_table_comment = ''
                # 表是否有分区
                partitioned = False
                if "`dt`" in create_table_str:
                    partitioned = True
                # 获取表列名
                hive_columns = spark_sql.get_columns(f'{hive_table_full_name}')

                column_type_flag = ['LONG', 'BIGINT', 'BOOLEAN', 'STRING', 'DOULBE']
                for column_name, column_value in hive_columns.items():
                    hive_column_names.append(column_name)
                    column_type = column_value[0].upper()
                    if column_type in column_type_flag:
                        hive_column_types[column_name] = column_type
                    else:
                        if column_type == 'INT':
                            hive_column_types[column_name] = 'LONG'

                kafka_writer_column = hive_column_names
                kafka_column_types = hive_column_types

            hdfs_reader_str = HDFSReader.generate_definition(hdfs_ds_name, hdfs_path, hive_database, hive_table_name,
                                                             hive_table_comment,
                                                             partitioned,
                                                             hive_column_names, hive_column_types)

            source_name = hive_table_name.replace('es_crl_', '').replace('_sum', '')
            kafka_writer_str = KafkaWriter.generate_definition(kafka_ds_name, kafka_topic, kafka_key, source_name,
                                                               kafka_writer_column, kafka_column_types,
                                                               hive_table_names_mapping)

            # 写文件
            generator_config_file = '{0}/hdfs-kafka-{1}-{2}.ini'.format(config_ini_path, hive_database, hive_table_name)
            write_file(f'{hdfs_reader_str}\n\n{kafka_writer_str}\n', generator_config_file)
            pretty_print(
                f'{NORM_YEL}{str(index).rjust(number_width, " ")}. {NORM_MGT} 读 Hive 数据库表 '
                f'{NORM_GRN}{hive_database}.{hive_table_name}'
                f'{NORM_MGT}\n                    写 Kafka DataX作业配置文件生成器配置已写入文件 '
                f'{NORM_GRN}{generator_config_file}'
            )
            index += 1


def hdfs_mongo_generator():
    hive_database = CONFIG.get('d')
    if not hive_database:
        pretty_print(f'{NORM_RED}参数 {NORM_GRN}-d{DO_RESET}{NORM_RED} 未提供')
        usage(1)
    included_tables = CONFIG.get('t', [])
    if isinstance(included_tables, str):
        included_tables = [included_tables]
    excluded_tables = CONFIG.get('e', [])
    if isinstance(excluded_tables, str):
        excluded_tables = [excluded_tables]
    if len(included_tables) == 0 and len(excluded_tables) == 0:
        pretty_print(f'{NORM_YEL}注意：'
                     f'{NORM_MGT}参数 {NORM_GRN}-t{NORM_MGT} 或 {NORM_GRN}-e '
                     f'{NORM_MGT}未提供，将扫描数据库 '
                     f'{NORM_GRN}{hive_database}{DO_RESET} '
                     f'{NORM_MGT}下所有表')
    with SparkSQL('datax-gc-generator') as spark_sql:
        # 获取json文件中表的主键
        table_pk_fields = []
        for validation_config_file in list_files('conf/validation', True):
            if validation_config_file.endswith('.json'):
                validation_config = load_json_file(validation_config_file)
                dwd_table_config = validation_config.get('dwd_table')  # type: str
                if not dwd_table_config:
                    pretty_print(f'{NORM_YEL}文件 {NORM_GRN}{validation_config_file}'
                                 f'{NORM_YEL} 中没有发现 {NORM_GRN}dwd_table{NORM_YEL} 的定义')
                    continue
                # 表名后缀，如：court_announcement
                table_name = dwd_table_config.split("_crl_")[1]
                ods_dwd_config = validation_config.get('ods_dwd_config')
                if not ods_dwd_config:
                    continue
                # 获取主键名
                new_pk_fields = ods_dwd_config.get('pk_fields')  # type:list
                # 放入到字典中
                for pk_fields in new_pk_fields:
                    table_pk_fields.append(table_name + "-" + pk_fields)
        hdfs_ds_name = 'hdfs-aliyun-cloud'
        hdfs_path = '/user/hive/warehouse'
        output = CONFIG.get('output', f'{base_dir}/ignored')
        config_ini_path = '{0}/config/hdfs-mongo/{1}'.format(output, hive_database)
        os.system(f'mkdir -p {config_ini_path}')
        final_tables = []
        if len(included_tables) == 0:
            all_hive_tables = []
            # 如果没有传入仅需的表则获取数据库下所有表
            tables_df, _ = spark_sql.query(f'show tables in {hive_database}', silent=True)
            desc_tables = tables_df.collect()
            for row in desc_tables:
                all_hive_tables.append(row[1])
            # 去除忽略的数据表
            for hive_table in all_hive_tables:
                if hive_table not in excluded_tables:
                    final_tables.append(f'{hive_database}.{hive_table}')
        else:
            final_tables = included_tables
        number_width = len(str(len(final_tables)))
        index = 1
        for hive_table_full_name in final_tables:
            hive_database, hive_table_name = hive_table_full_name.split('.')
            # 获取表备注
            create_table_str = spark_sql.query_scalar(f'show create table {hive_table_full_name}', silent=True)
            hive_table_comment = create_table_str.split('ROW FORMAT SERDE')[0].split(")")[1]
            if 'COMMENT' in hive_table_comment:
                hive_table_comment = hive_table_comment.split("'")[1]
            else:
                hive_table_comment = ''
            # 表是否有分区
            partitioned = False
            if "`dt`" in create_table_str:
                partitioned = True
            # 获取表列名
            hive_columns = spark_sql.get_columns(f'{hive_table_full_name}')
            hive_column_names = []
            hive_column_types = {}
            for column_name, column_value in hive_columns.items():
                hive_column_names.append(column_name)
                column_type = column_value[0].upper()
                if column_type == 'BIGINT':
                    column_type = 'LONG'
                hive_column_types[column_name] = column_type
            hdfs_reader_str = HDFSReader.generate_definition(hdfs_ds_name, hdfs_path, hive_database, hive_table_name,
                                                             hive_table_comment,
                                                             partitioned,
                                                             hive_column_names, hive_column_types)
            mongo_ds_name = 'mongo-dev-rw'
            if hive_table_full_name.__contains__('_crl_'):
                split_table_name = hive_table_full_name.split("_crl_")[1]
            else:
                split_table_name = 'unknown'
            mongo_database = 'enterprise'
            mongo_collection = snake_case_to_pascal_case(split_table_name)
            # mongo列名
            mongo_column_names = []
            for column_name in hive_column_names:
                if MONGO_SPECIAL_WORDS_DICT.__contains__(column_name):
                    mongo_column_names.append(MONGO_SPECIAL_WORDS_DICT[column_name])
                else:
                    column_name = snake_case_to_camel_case(column_name)
                    mongo_column_names.append(column_name)
            # mongo列类型和主键
            mongo_column_types = {}
            pk_fields = []
            for column_name, column_type in hive_column_types.items():
                is_pk = False
                if split_table_name + "-" + column_name in table_pk_fields:
                    is_pk = True
                if MONGO_SPECIAL_WORDS_DICT.__contains__(column_name):
                    column_name = MONGO_SPECIAL_WORDS_DICT[column_name]
                else:
                    column_name = snake_case_to_camel_case(column_name)
                if column_name == 'createDate' and column_type == 'LONG':
                    mongo_column_types[column_name] = 'DATE'
                elif column_name == 'updateDate' and column_type == 'LONG':
                    mongo_column_types[column_name] = 'DATE'
                else:
                    mongo_column_types[column_name] = column_type
                if is_pk:
                    pk_fields.append(column_name)

            mongo_writer_str = MongoWriter.generate_definition(mongo_ds_name, mongo_database, mongo_collection,
                                                               mongo_column_names,
                                                               mongo_column_types, pk_fields)
            # 写文件
            generator_config_file = '{0}/hdfs-mongo-{1}-{2}.ini'.format(config_ini_path, mongo_database, split_table_name)
            write_file(f'{hdfs_reader_str}\n\n{mongo_writer_str}\n', generator_config_file)
            pretty_print(
                f'{NORM_YEL}{str(index).rjust(number_width, " ")}. {NORM_MGT} 读 Hive 数据库表 '
                f'{NORM_GRN}{hive_database}.{hive_table_name}'
                f'{NORM_MGT}\n                    写 MongoDB DataX作业配置文件生成器配置已写入文件 '
                f'{NORM_GRN}{generator_config_file}'
            )
            index += 1


def hdfs_mysql_generator():
    raise Exception(f'Not implemented yet with from {from_system} and to {to_system}')


def mysql_hdfs_generator():
    host = CONFIG.get('h')
    port = int(CONFIG.get('P', '3306'))
    username = CONFIG.get('u')
    password = CONFIG.get('p')
    mysql_database = CONFIG.get('D')
    tables = CONFIG.get('t', [])
    if isinstance(tables, str):
        tables = [tables]
    excluded_tables = CONFIG.get('e', [])
    if isinstance(excluded_tables, str):
        excluded_tables = [excluded_tables]
    table_regex = CONFIG.get('tr', [])
    if isinstance(table_regex, str):
        table_regex = [table_regex]
    exclude_regex = CONFIG.get('er', [])
    if isinstance(exclude_regex, str):
        exclude_regex = [exclude_regex]
    project = CONFIG.get('project')
    layer = CONFIG.get('layer', 'ods')
    edition = CONFIG.get('edition')
    env = CONFIG.get('env')
    partitioned = CONFIG.get('partitioned', False)
    inc_col = CONFIG.get('inc-col', 'update_time')
    if not (host and username and password and mysql_database):
        usage(1)
    hdfs_ds_name = 'hdfs-aliyun-cloud'
    hdfs_default_fs = 'hdfs://cluster'
    hdfs_path = '/user/hive/warehouse'
    output = CONFIG.get('output', f'{base_dir}/ignored')
    hive_database = get_hive_database_name(project, layer, env)
    hive_table_prefix = get_hive_table_prefix(project, layer, edition)
    mysql_datasource_path = '{0}/datasource/mysql/{1}'.format(output, hive_database)
    hdfs_datasource_path = '{0}/datasource/hdfs'.format(output)
    config_ini_path = '{0}/config/mysql-hdfs/{1}'.format(output, hive_database)
    hive_ddl_path = '{0}/ddl'.format(output)
    hive_ddl_file = '{0}/{1}.sql'.format(hive_ddl_path, mysql_database)
    os.system(f'mkdir -p {mysql_datasource_path}')
    os.system(f'mkdir -p {hdfs_datasource_path}')
    os.system(f'mkdir -p {config_ini_path}')
    os.system(f'mkdir -p {hive_ddl_path}')
    mysql_ds_def = MySQLDataSource.generate_definition(host, port, username, password, mysql_database)
    hdfs_ds_def = HDFSDataSource.generate_definition(hdfs_default_fs)
    mysql_handler = MySQLHandler(host, port, username, password)
    mysql_tables = mysql_handler.list_tables(mysql_database, exclude_regex, table_regex)
    number_width = len(str(len(mysql_tables)))
    index = 1
    if os.path.exists(hive_ddl_file):
        hive_ddl_file = '{0}/{1}-{2}.sql'.format(hive_ddl_path, mysql_database, str(int(time.time())))
    write_file('CREATE DATABASE IF NOT EXISTS %s;\n' % hive_database, hive_ddl_file)
    for mysql_table_name, mysql_table_comment in mysql_tables.items():
        if tables and not tables.__contains__(mysql_table_name):
            continue
        if excluded_tables and excluded_tables.__contains__(mysql_table_name):
            continue
        mysql_column_list = mysql_handler.list_columns(mysql_database, mysql_table_name)
        mysql_column_names = [c.COLUMN_NAME for c in mysql_column_list]
        column_types = convert_mysql_column_types(mysql_column_list)
        mysql_reader_def = MySQLReader.generate_definition(
            mysql_database,
            mysql_table_name, mysql_table_comment,
            mysql_column_names, column_types,
            hive_database, partitioned, inc_col
        )

        hive_table_name = f'{hive_table_prefix}_{mysql_table_name}'

        hive_ddl_def = MySQLReader.generate_hive_ddl(
            hive_database, hive_table_name, mysql_table_comment,
            partitioned,
            mysql_column_list, column_types
        )
        hdfs_writer_def = HDFSWriter.generate_definition(
            hdfs_ds_name, hdfs_path,
            hive_database, hive_table_name, partitioned,
            mysql_column_names, column_types
        )
        generator_config_file = '{0}/mysql-hdfs-{1}-{2}.ini'.format(config_ini_path, mysql_database, mysql_table_name)
        write_file(f'{mysql_reader_def}\n{hdfs_writer_def}\n', generator_config_file)
        append_file(f'{hive_ddl_def}\n', hive_ddl_file)
        pretty_print(
            f'{NORM_YEL}{str(index).rjust(number_width, " ")}{NORM_MGT} 读 MySQL 数据库表 '
            f'{NORM_GRN}{mysql_database}.{mysql_table_name}'
            f'{NORM_MGT}\n                    写 HDFS DataX作业配置文件生成器配置已写入文件 '
            f'{NORM_GRN}{generator_config_file}'
            f'{NORM_MGT}\n                    对应的Hive建表DDL已写入文件 '
            f'{NORM_GRN}{hive_ddl_file}'
        )
        index += 1
        # break
    mysql_ds_def_file = '{0}/mysql-{1}.ini'.format(mysql_datasource_path, mysql_database)
    if os.path.exists(mysql_ds_def_file):
        mysql_ds_def_file = '{0}/mysql-{1}-{2}.ini'.format(mysql_datasource_path, mysql_database, str(int(time.time())))
    write_file(f'{mysql_ds_def}\n', mysql_ds_def_file)
    pretty_print(f'{NORM_MGT}MySQL数据库 {NORM_GRN}{mysql_database}{NORM_MGT} 数据源配置已写入文件 {NORM_GRN}{mysql_ds_def_file}')
    hdfs_ds_def_file = '{0}/{1}.ini'.format(hdfs_datasource_path, hdfs_ds_name)
    if os.path.exists(hdfs_ds_def_file):
        hdfs_ds_def_file = '{0}/{1}-{2}.ini'.format(hdfs_datasource_path, hdfs_ds_name, str(int(time.time())))
    write_file(f'{hdfs_ds_def}\n', hdfs_ds_def_file)
    pretty_print(f'{NORM_MGT}HDFS 数据源配置已写入文件 {NORM_GRN}{hdfs_ds_def_file}')


def mysql_hbase_generator():
    host = CONFIG.get('h')
    port = int(CONFIG.get('P', '3306'))
    username = CONFIG.get('u')
    password = CONFIG.get('p')
    mysql_database = CONFIG.get('D')
    if not (host and username and password and mysql_database):
        usage(1)
    included_tables = CONFIG.get('t', [])
    if isinstance(included_tables, str):
        included_tables = [included_tables]
    excluded_tables = CONFIG.get('e', [])
    if isinstance(excluded_tables, str):
        excluded_tables = [excluded_tables]
    table_regex = CONFIG.get('tr', [])
    if isinstance(table_regex, str):
        table_regex = [table_regex]
    exclude_regex = CONFIG.get('er', [])
    if isinstance(exclude_regex, str):
        exclude_regex = [exclude_regex]
    if len(included_tables) == 0 and len(excluded_tables) == 0 and len(table_regex) == 0 and len(exclude_regex) == 0:
        pretty_print(f'{NORM_YEL}注意：'
                     f'{NORM_MGT}参数 {NORM_GRN}-t -e -tr -er '
                     f'{NORM_MGT}都未提供，将扫描数据库 '
                     f'{NORM_GRN}{mysql_database}{DO_RESET} '
                     f'{NORM_MGT}下所有表')
    hbase_namespace = CONFIG.get('n')
    if not hbase_namespace:
        pretty_print(f'{NORM_YEL}注意：'
                     f'{NORM_MGT}参数 {NORM_GRN}-n{DO_RESET} '
                     f'{NORM_MGT}未提供，将使用 HBase 默认命名空间 '
                     f'{NORM_GRN}default')
        hbase_namespace = 'default'
    project = CONFIG.get('project')
    layer = CONFIG.get('layer', 'ods')
    edition = CONFIG.get('edition')
    env = CONFIG.get('env')
    partitioned = CONFIG.get('partitioned', False)
    inc_col = CONFIG.get('inc-col', 'update_time')
    hbase_ds_name = 'hbase-default'
    hdfs_default_fs = 'hdfs://cluster'
    output = CONFIG.get('output', f'{base_dir}/ignored')
    hive_database = get_hive_database_name(project, layer, env)
    hive_table_prefix = get_hive_table_prefix(project, layer, edition)
    mysql_datasource_path = '{0}/datasource/mysql/{1}'.format(output, hive_database)
    config_ini_path = '{0}/config/mysql-hbase/{1}'.format(output, hive_database)
    hive_ddl_path = '{0}/ddl'.format(output)
    hive_ddl_file = '{0}/{1}.sql'.format(hive_ddl_path, mysql_database)
    os.system(f'mkdir -p {mysql_datasource_path}')
    os.system(f'mkdir -p {config_ini_path}')
    os.system(f'mkdir -p {hive_ddl_path}')
    mysql_ds_def = MySQLDataSource.generate_definition(host, port, username, password, mysql_database)
    mysql_handler = MySQLHandler(host, port, username, password)
    mysql_tables = mysql_handler.list_tables(mysql_database, exclude_regex, table_regex)
    number_width = len(str(len(mysql_tables)))
    index = 1
    if os.path.exists(hive_ddl_file):
        hive_ddl_file = '{0}/{1}-{2}.sql'.format(hive_ddl_path, mysql_database, str(int(time.time())))
    write_file('CREATE DATABASE IF NOT EXISTS %s;\n' % hive_database, hive_ddl_file)
    for mysql_table_name, mysql_table_comment in mysql_tables.items():
        if included_tables and not included_tables.__contains__(mysql_table_name):
            continue
        if excluded_tables and excluded_tables.__contains__(mysql_table_name):
            continue
        mysql_column_list = mysql_handler.list_columns(mysql_database, mysql_table_name)
        mysql_column_names = [c.COLUMN_NAME for c in mysql_column_list]
        column_types = convert_mysql_column_types(mysql_column_list)
        mysql_reader_def = MySQLReader.generate_definition(
            mysql_database,
            mysql_table_name, mysql_table_comment,
            mysql_column_names, column_types,
            hive_database, partitioned, inc_col
        )

        hive_table_name = f'{hive_table_prefix}_{mysql_table_name.lower()}_hbase_mapping'
        hbase_table_name = mysql_table_name.lower()

        hive_over_hbase_ddl_def = MySQLReader.generate_hive_over_hbase_ddl(
            hive_database, hive_table_name, mysql_table_comment, hbase_namespace, hbase_table_name,
            mysql_column_list, column_types
        )
        row_key_columns = []
        row_key_columns.append('reverse(主键如果是自增ID，建议reverse)')
        row_key_columns.append('separator(@@)')
        row_key_columns.append(f'separator({mysql_database})')
        row_key_columns.append(f'separator(.)')
        row_key_columns.append(f'separator({mysql_table_name})')
        hdfs_writer_def = HBaseWriter.generate_definition(
            hbase_ds_name, hbase_namespace, hbase_table_name,
            mysql_table_name, mysql_table_comment, 'cf',
            mysql_column_names, column_types, row_key_columns
        )
        generator_config_file = '{0}/mysql-hbase-{1}-{2}.ini'.format(config_ini_path, mysql_database, mysql_table_name)
        write_file(f'{mysql_reader_def}\n{hdfs_writer_def}\n', generator_config_file)
        append_file(f'{hive_over_hbase_ddl_def}\n', hive_ddl_file)
        pretty_print(
            f'{NORM_YEL}{str(index).rjust(number_width, " ")}. {NORM_MGT} 读 MySQL 数据库表 '
            f'{NORM_GRN}{mysql_database}.{mysql_table_name}'
            f'{NORM_MGT}\n                    写 HBase DataX作业配置文件生成器配置已写入文件 '
            f'{NORM_GRN}{generator_config_file}'
            f'{NORM_MGT}\n                    对应的 Hive Over HBase 建表DDL已写入文件 '
            f'{NORM_GRN}{hive_ddl_file}'
        )
        index += 1
        # break
    mysql_ds_def_file = '{0}/mysql-{1}.ini'.format(mysql_datasource_path, mysql_database)
    if os.path.exists(mysql_ds_def_file):
        mysql_ds_def_file = '{0}/mysql-{1}-{2}.ini'.format(mysql_datasource_path, mysql_database, str(int(time.time())))
    write_file(f'{mysql_ds_def}\n', mysql_ds_def_file)
    pretty_print(f'{NORM_MGT}MySQL数据库 {NORM_GRN}{mysql_database}{NORM_MGT} 数据源配置已写入文件 {NORM_GRN}{mysql_ds_def_file}')


if __name__ == '__main__':
    pretty_print(f'{NORM_MGT}{sys.argv[0]} 收到参数：{NORM_GRN}{" ".join(sys.argv[1:])}')
    CONFIG, _ = parse_args(sys.argv[1:])
    # 未提供任何参数或查看帮助
    if exist(CONFIG, ['H', 'help']):
        usage(0)
    from_system = CONFIG.get('from')
    if from_system:
        from_system = from_system.lower()
    to_system = CONFIG.get('to')
    if to_system:
        to_system = to_system.lower()
    if not from_system or not to_system:
        usage(1)
    if from_system == "hdfs":
        if to_system == "elasticsearch":
            hdfs_elasticsearch_generator()
        elif to_system == "hbase":
            hdfs_hbase_generator()
        elif to_system == "kafka":
            hdfs_kafka_generator()
        elif to_system == "mongo":
            hdfs_mongo_generator()
        elif to_system == "mysql":
            hdfs_mysql_generator()
        else:
            raise Exception(f'Not implemented yet with from {from_system} and to {to_system}')
    elif from_system == "mysql":
        if to_system == "hbase":
            mysql_hbase_generator()
        elif to_system == "hdfs":
            mysql_hdfs_generator()
        else:
            raise Exception(f'Not implemented yet with from {from_system} and to {to_system}')
    else:
        raise Exception(f'Not implemented yet with from {from_system} and to {to_system}')