python結(jié)合shell自動創(chuàng)建kafka的連接器實戰(zhàn)教程

第python結(jié)合shell自動創(chuàng)建kafka的連接器實戰(zhàn)教程目錄環(huán)境安裝連接oracle的python包獲取oracle表信息去掉多余部分shell腳本處理表信息文件增強版處理表信息腳本

環(huán)境

cat/etc/redhat-release

CentOSLinuxrelease7.5.1804(Core)

[root@localhost~]#uname-a

Linuxlocalhost.localdomain3.10.0-862.el7.x86_64#1SMPFriApr2016:44:24UTC2025x86_64x86_64x86_64GNU/Linux

python-V

Python2.7.5

安裝連接oracle的python包

pipinstallcx_Oracle==7.3

獲取oracle表信息

catquery_oracle.py

#!/usr/bin/envpython

importcx_Oracle

importsys

importos

importcsv

importtraceback

file=open("oracle.txt",'w').close()

user="test"

passwd="test"

listener='5:1521/orcl'

conn=cx_Oracle.connect(user,passwd,listener)

cursor=conn.cursor()

sql="selecttable_namefromuser_tables"

cursor.execute(sql)

LIST1=[]

whileTrue:

row=cursor.fetchone()

ifrow==None:

break

fortableinrow:

#printtable

LIST1.append(table)

LIST2=[]

foriinLIST1:

sql3="selectCOLUMN_NAME,DATA_TYPE,DATA_PRECISION,DATA_SCALEfromcolsWHERETABLE_name=upper('%s')"%i

cursor.execute(sql3)

cursor.execute(sql3)

row3=cursor.fetchall()

fordatainrow3:

#LIST2.append(i)

LIST2.extend(list(data))

LIST2.append(i)

f=open('oracle.txt','a+')

printf,LIST2

LIST2=[]

#f=open('test.txt','a+')

#selecttable_name,column_name,DATA_TYPEfromcolsWHERETABLE_name=upper('student');

#selectcolumn_name,DATA_TYPEfromcolsWHERETABLE_name=upper('student');

去掉多余部分

catauto.sh

#!/bin/bash

#pythonquery_oracle.py|tr","''|tr"'"''|tr"["""|tr"]"""

#oracle.txt

oracle_tables.txt

catoracle.txt|tr"[],'"""|sed"s#[][]*##g"|sed's/^[\t]*//g'oracle_tables.txt

catoracle_tables.txt

SNONUMBER190SNAMEVARCHAR2NoneNoneSSEXVARCHAR2NoneNoneSBIRTHDAYDATENoneNoneSCLASSVARCHAR2NoneNoneSTUDENTDATE_DATE

SNO2NUMBER190SNAMEVARCHAR2NoneNoneSSEXVARCHAR2NoneNoneSBIRTHDAYDATENoneNoneSCLASSVARCHAR2NoneNoneSTUDENT2INPUT_TIME

SNO3NUMBER192SNAMEVARCHAR2NoneNoneSSEXVARCHAR2NoneNoneSBIRTHDAYDATENoneNoneSCLASSVARCHAR2NoneNoneSTUDENT3DATA_DATE

shell腳本處理表信息文件

catconnect.sh

#!/bin/bash

#獲取臨時文件的行數(shù)

FILE_NUM=$(catoracle_tables.txt|egrep-v'#|^$'|wc-l)

#清空自動創(chuàng)建連接器的腳本

create-connect.sh

#循環(huán)臨時文件每一行

foriin`seq$FILE_NUM`

FILE_LINE=$(sed-n${i}poracle_tables.txt)

TABLE_NAME=$(echo${FILE_LINE}|sed's/[\t]*$//g'|awk'{print$(NF-1)}')

COL_NUM=$(echo${FILE_LINE}|sed's/[\t]*$//g'|awk-F"[]"'{printNF}')

REAL_COL_NUM=`expr$COL_NUM-2`

#清空臨時文件

${TABLE_NAME}.txt

${TABLE_NAME}.sql

#循環(huán)臨時文件每行列名所在的列

forjin`seq14$REAL_COL_NUM`

k=`expr$j+1`

m=`expr$j+2`

n=`expr$j+3`

COL_NAME=$(echo$FILE_LINE|cut-d""-f${j})

COL_DATA_TYPE=$(echo$FILE_LINE|cut-d""-f${k})

COL_DATA_PRECISION=$(echo$FILE_LINE|cut-d""-f${m})

COL_DATA_SCALE=$(echo$FILE_LINE|cut-d""-f${n})

#判斷列的數(shù)據(jù)類型是否是NUMBER

if["$COL_DATA_TYPE"="NUMBER"]

then

#循環(huán)拼接SQL查詢中的CAST(*AS*)AS*部分,追加到臨時文件中

echo"CAST($COL_NAMEAS$COL_DATA_TYPE($COL_DATA_PRECISION,$COL_DATA_SCALE))AS$COL_NAME"${TABLE_NAME}.txt

else

#循環(huán)拼接SQL查詢中的列名部分，追加到臨時文件中

echo"$COL_NAME"${TABLE_NAME}.txt

done

#拼接完整的SQL語句,追加到臨時文件中

echo"select$(cat${TABLE_NAME}.txt|tr"\n"","|sed-e's/,$/\n/')from$TABLE_NAMEwhere$(sed-n${i}poracle_tables.txt|cut-d''-f$COL_NUM)=trunc(sysdate-2)and$(sed-n${i}poracle_tables.txt|cut-d''-f$COL_NUM)trunc(sysdate-1)"${TABLE_NAME}.sql

#循環(huán)追加每個表對應(yīng)的連接器到自動創(chuàng)建連接器的腳本中

catcreate-connect.shEOF

curl-XPOSThttp://localhost:8083/connectors-H"Content-Type:application/json"-d'{

"name":"jdbc_source_$TABLE_NAME",

"config":{

"connector.class":"io.confluent.connect.jdbc.JdbcSourceConnector",

"connection.url":"jdbc:oracle:thin:@{{ORACLE_IP}}:{{ORACLE_PORT}}:orcl",

"connection.user":"{{ORACLE_USER}}",

"connection.password":"{{ORACLE_PASSWD}}",

"topic.prefix":"YC_$TABLE_NAME",

"mode":"{{CONNECT_MODE}}",

"query":"$(cat${TABLE_NAME}.sql)"

}'/dev/null21

說明：腳本中{{變量名}}部分的內(nèi)容是獲取ansible中的變量，這個腳本是和ansible結(jié)合使用的。

增強版處理表信息腳本

#!/bin/bash

#獲取臨時文件的行數(shù)

FILE_NUM=$(catoracle_time_tables.txt|egrep-v'#|^$'|wc-l)

#清空創(chuàng)建連接器的腳本并追加echos函數(shù)

create-jdbc-connect.sh

catcreate-jdbc-connect.shEOF

#!/bin/bash

echos(){

case\$1in

red)echo-e"\033[31m\$2\033[0m";;

green)echo-e"\033[32m\$2\033[0m";;

yellow)echo-e"\033[33m\$2\033[0m";;

blue)echo-e"\033[34m\$2\033[0m";;

purple)echo-e"\033[35m\$2\033[0m";;

*)echo"\$2";;

create-jdbc-connect-time.sh

catcreate-jdbc-connect-time.shEOF

#!/bin/bash

echos(){

case\$1in

red)echo-e"\033[31m\$2\033[0m";;

green)echo-e"\033[32m\$2\033[0m";;

yellow)echo-e"\033[33m\$2\033[0m";;

blue)echo-e"\033[34m\$2\033[0m";;

purple)echo-e"\033[35m\$2\033[0m";;

*)echo"\$2";;

#創(chuàng)建表相關(guān)文件目錄

mkdir-p./TABLE_TIME

#循環(huán)臨時文件每一行

foriin`seq$FILE_NUM`

FILE_LINE=$(sed-n${i}poracle_time_tables.txt)

TABLE_NAME=$(echo${FILE_LINE}|sed's/[\t]*$//g'|awk'{print$(NF)}')

COL_NUM=$(echo${FILE_LINE}|sed's/[\t]*$//g'|awk-F"[]"'{printNF}')

REAL_COL_NUM=`expr$COL_NUM-2`

#清空臨時文件

./TABLE_TIME/${TABLE_NAME}_time.txt

./TABLE_TIME/${TABLE_NAME}_time.sql

./TABLE_TIME/${TABLE_NAME}.sql

#循環(huán)臨時文件每行列名所在的列

forjin`seq14$REAL_COL_NUM`

k=`expr$j+1`

m=`expr$j+2`

n=`expr$j+3`

COL_NAME=$(echo$FILE_LINE|cut-d""-f${j})

COL_DATA_TYPE=$(echo$FILE_LINE|cut-d""-f${k})

COL_DATA_PRECISION=$(echo$FILE_LINE|cut-d""-f${m})

COL_DATA_SCALE=$(echo$FILE_LINE|cut-d""-f${n})

#判斷列的數(shù)據(jù)類型是否是NUMBER

if["$COL_DATA_TYPE"="NUMBER"]

then

#循環(huán)拼接SQL查詢中的CAST(*AS*)AS*部分,追加到臨時文件中

echo"CAST($COL_NAMEAS$COL_DATA_TYPE($COL_DATA_PRECISION,$COL_DATA_SCALE))AS$COL_NAME"./TABLE_TIME/${TABLE_NAME}_time.txt

else

#循環(huán)拼接SQL查詢中的列名部分，追加到臨時文件中

echo"$COL_NAME"./TABLE_TIME/${TABLE_NAME}_time.txt

#判斷是否存在hosts中定義的時間列，如果有就追加該列名進一個臨時文件中

TIME_COL=({{TABLE_TIME_COL}})

forTIMEin${TIME_COL[@]}

if["$COL_NAME"="$TIME"]

then

echo"$COL_NAME"./TABLE_TIME/${TABLE_NAME}_TIME_COL.txt

done

done

#拼接完整的SQL語句,追加到臨時文件中

if[-f"./TABLE_TIME/${TABLE_NAME}_TIME_COL.txt"]

then

#echo"select$(cat./TABLE_TIME/${TABLE_NAME}.txt|tr"\n"","|sed-e's/,$/\n/')from{{ORACLE_TABLES_USER}}.$TABLE_NAMEwhere$(sed-n${i}poracle_tables.txt|cut-d''-f$COL_NUM)=trunc(sysdate-2)and$(sed-n${i}poracle_tables.txt|cut-d''-f$COL_NUM)trunc(sysdate-1)"./TABLE_TIME/${TABLE_NAME}_time.sql

echo"select$(cat./TABLE_TIME/${TABLE_NAME}_time.txt|tr"\n"","|sed-e's/,$/\n/')from{{ORACLE_TABLES_USER}}.$TABLE_NAMEwhere$(cat./TABLE_TIME/${TABLE_NAME}_TIME_COL.txt)=trunc(sysdate-2)and$(cat./TABLE_TIME/${TABLE_NAME}_TIME_COL.txt)trunc(sysdate-1)"./TABLE_TIME/${TABLE_NAME}_time.sql

else

echo"select$(cat./TABLE_TIME/${TABLE_NAME}_time.txt|tr"\n"","|sed-e's/,$/\n/')from{{ORACLE_TABLES_USER}}.$TABLE_NAME"./TABLE_TIME/${TABLE_NAME}.sql

#循環(huán)追加每個表對應(yīng)的連接器到自動創(chuàng)建連接器的腳本中

if[-f"./TABLE_TIME/${TABLE_NAME}_TIME_COL.txt"]

catcreate-jdbc-connect-time.shEOF

#創(chuàng)建表$TABLE_NAME連接器的命令如下

curl-XPOSThttp://localhost:8083/connectors-H"Content-Type:application/json"-d'{

"name":"jdbc_time_$TABLE_NAME",

"config":{

"connector.class":"io.confluent.connect.jdbc.JdbcSourceConnector",

"connection.url":"jdbc:oracle:thin:@{{ORACLE_IP}}:{{ORACLE_PORT}}:{{ORACLE_SERVER_NAME}}",

"connection.user":"{{ORACLE_USER}}",

"connection.password":"{{ORACLE_PASSWD}}",

"topic.prefix":"YC_${TABLE_NAME}_INSERT",

"erval.ms":"86400000",

"mode":"{{CONNECT_MODE}}",

"numeric.mapping":"best_fit",

"query":"$(cat./TABLE_TIME/${TABLE_NAME}_time.sql)"

}'/dev/null21

#判斷連接器是否創(chuàng)建成功

if[\$-eq0]

echosgreen"\$(date+"%F%H:%M:%S")創(chuàng)建jdbc_time_${TABLE_NAME}連