python小脚本-python自动生成oracle

需求说明：

许多表的情况下，使用Python自动生成从Oracle到hdfs的sqoop脚本。

思路：

1.将需要使用sqoop抽数的表名字存放在tables.txt文件中；

2.Python读取tables.txt中的表名，生成可以去Oracle执行的sql语句来查询表字段：

 ；

3.拼接这些字段和sqoop命令生成抽取Oracle到hdfs的shell脚本。

需要sqoop抽数的表名字：

 Python生成的sqoop的shell脚本：

import cx_Oracle as co
import os

#获取Oracle的连接，并打开游标
def conect_oracle_open():
    global connect
    connect=co.connect('用户名', '密码', 'IP:端口/实例名')
    global cursor
    cursor=connect.cursor()

#关闭orale的连接，先关游标，再关连接
def conect_oracle_close():
    cursor.close()
    connect.close()

#从table.txt表中获取去Oracle执行的sql语句
#返回表名|能获取表字段的sql语句
#ATM_TAB_USER|select column_name ,data_type from user_tab_cols where table_name = 'ATM_TAB_USER'
def read_tables_to_get_sql():
    with open(r"C:UsersDesktoptesttables.txt", encoding='utf8') as f:
        table_lists = f.readlines()  # 列表
    sql_lists = []
    for table in table_lists:
        tmp = table.replace('n', '').split()[0]
        sql = f"select column_name ,data_type from user_tab_cols where table_name = '{tmp}'"
        tname_sql=tmp+"|"+sql
        #print(tname_sql)
        #ATM_TAB_USER|select column_name ,data_type from user_tab_cols where table_name = 'ATM_TAB_USER'
        sql_lists.append(tname_sql)
    return sql_lists

#拼接sqoop参数 并返回完整的sqoop命令，--query后面的select语句和表名字作为参数
#hadoop fs -rm -r -f /tmp/date_$in_name/SOR/ATM_TAB_TRANSACTION_INFOR
#sqoop import --connect jdbc:oracle:thin:@//$source_ip:$source_port/$source_database --username $USERNAME --password $PASSWORD --query "select TI_TIME,BCI_CARD_NUM,TI_TYPE,TI_MONEY,TI_RECORD from ATM_TAB_TRANSACTION_INFOR where $CONDITIONS " --target-dir /tmp/khzb/date_$in_name/SOR/CNL_USR_STS_INFO -m 1 --fields-terminated-by '\001'--hive-drop-import-delims --input-null-string '\N' --input-null-non-string '\N'
def command_concatenate(sql_select,table_name):
    support_retry = "hadoop fs -rm -r -f /tmp/date_$in_name/SOR/" + table_name
    sqoop_pre = "sqoop import --connect jdbc:oracle:thin:@//$source_ip:$source_port/$source_database --username $USERNAME --password $PASSWORD --query "
    sqoop_sql = ""select" + " " + sql_select + " from " + table_name + " where $CONDITIONS " + """
    sqoop_last = " --target-dir /tmp/khzb/date_$in_name/SOR/CNL_USR_STS_INFO -m 1 --fields-terminated-by " + str(
        r"'\001'") + "--hive-drop-import-delims --input-null-string " + r"'\N'" + " --input-null-non-string " + r"'\N'"
    sqoop_command = support_retry + "n" + sqoop_pre + sqoop_sql + sqoop_last
    return sqoop_command

#去Oracle执行sql，获取每一张表的sqoop命令
#返回数组，数组元素是所有表的sqoop命令
def execute_sql_to_get_table_structure():
    sqoop_commands = []
    sql_lists = read_tables_to_get_sql() #表名字|能去Oracle查询字段的sql
    conect_oracle_open()
    for sql_stru in sql_lists:
        #print("555555555555555" + sql_stru.split('|')[0])
        cursor.execute(sql_stru.split('|')[1])
        rows = cursor.fetchall()
        sql_select = ''
        for row in rows:
            if len(sql_select) == 0:
                sql_select = row[0]
                # print(sql_select)
            else:
                sql_select = sql_select + "," + row[0]
        # 已经获取到--query后面的select语句
        table_name=sql_stru.split('|')[0]#解析表名字
        #print("table::::::"+table_name)
        sqoop_command = command_concatenate(sql_select,table_name)
        sqoop_commands.append(sqoop_command)
    return sqoop_commands


def save_shell_script():
    if os.path.exists(r'C:UsersDesktoptest2.txt') :
        os.remove(r'C:UsersDesktoptest2.txt')
        print("the file exists ,now remove")
    else:
        print("the file not exists,now will create!")
    sqoop_commands=execute_sql_to_get_table_structure()
    for sqoop_command in sqoop_commands:
        with open(r'C:UsersDesktoptest2.txt', 'a', encoding='utf8') as f:
            f.write(sqoop_command+"nn")

def main():
    save_shell_script()
    conect_oracle_close()

if __name__ == "__main__":
    main()