85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
import sqlparse
|
||
import sqlglot
|
||
from sqlglot.expressions import ColumnDef
|
||
from utils.log import Log
|
||
|
||
def odps(schema,table_name,columns,colmapping,hologres_connection):
|
||
|
||
odps_sql=f'''
|
||
CREATE EXTERNAL TABLE IF NOT EXISTS {table_name}
|
||
(
|
||
{columns}
|
||
)
|
||
STORED BY 'com.aliyun.odps.jdbc.JdbcStorageHandler'
|
||
-- ip设置成经典网络ip 库 加Schema 加表名
|
||
location 'jdbc:postgresql://{hologres_connection}/{schema}?ApplicationName=MaxCompute¤tSchema={schema}&preferQueryMode=simple&useSSL=false&table={table_name}/'
|
||
TBLPROPERTIES (
|
||
'mcfed.mapreduce.jdbc.driver.class'='org.postgresql.Driver',
|
||
'odps.federation.jdbc.target.db.type'='holo',
|
||
-- 格式为：MaxCompute字段1 : "Hologres字段1",MaxCompute字段2 : "Hologres字段2"
|
||
'odps.federation.jdbc.colmapping'='{colmapping}'
|
||
);
|
||
'''
|
||
return odps_sql
|
||
|
||
def extract_create_table(sql_script):
|
||
# 创建一个新的Log实例，确保每天创建一个新的日志文件
|
||
log = Log().getlog()
|
||
# 解析 SQL 脚本
|
||
parsed = sqlparse.parse(sql_script)
|
||
|
||
create_table_statements = []
|
||
|
||
for statement in parsed:
|
||
# 关闭格式化选项保持原样
|
||
stripped = sqlparse.format(
|
||
statement.value,
|
||
strip_comments=True,
|
||
reindent=False,
|
||
keyword_case="lower"
|
||
)
|
||
|
||
# 跳过空语句
|
||
if not stripped.strip():
|
||
continue
|
||
|
||
# 可修改条件来匹配其他语句类型
|
||
if stripped.upper().strip().startswith(("CREATE TABLE")):
|
||
create_table_statements.append(stripped)
|
||
|
||
return "\n".join(create_table_statements)
|
||
|
||
def parse_create_table_sql(create_table_sql,hologres_connection):
|
||
# 创建一个新的Log实例，确保每天创建一个新的日志文件
|
||
log = Log().getlog()
|
||
|
||
result = extract_create_table(create_table_sql)
|
||
|
||
re_create_table_sql = sqlglot.transpile(result, read="postgres", write="hive")[0]
|
||
|
||
parsed = sqlglot.parse_one(re_create_table_sql, read='hive')
|
||
|
||
# 获取表名
|
||
table_name = parsed.this.this
|
||
|
||
columns = []
|
||
colmapping = []
|
||
# 遍历所有可能包含列定义的子表达式
|
||
for expression in parsed.walk():
|
||
if isinstance(expression[0], ColumnDef):
|
||
# 获取列名
|
||
column_name = expression[0].this.this
|
||
# 获取数据类型
|
||
column_type = expression[0].args['kind'].this.name.upper()
|
||
# 如果是TEXT类型，则转换为STRING
|
||
if column_type == 'TEXT':
|
||
column_type = 'STRING'
|
||
columns.append(column_name+" "+column_type)
|
||
colmapping.append(column_name+":"+column_name)
|
||
# 将columns,colmapping转换成字符串用,分割
|
||
columns_str = ",\n".join(columns)
|
||
colmapping_str = ",".join(colmapping)
|
||
table_name_str=str(table_name).split('.')[-1]
|
||
schema=str(table_name).split('.')[0]
|
||
|
||
return odps(schema,table_name_str,columns_str,colmapping_str,hologres_connection) |