本文目录导读：
我来详细说明如何用Python生成Impala表配置（包括建表语句和Hive元数据配置）。
生成Impala建表语句
基础建表语句生成
def generate_create_table_sql(table_name, columns, partition_cols=None,
storage_format='PARQUET', location=None):
"""
生成Impala建表SQL语句
Args:
table_name: 表名（可包含数据库名）
columns: 列定义列表 [('col1', 'STRING', 'comment1'), ...]
partition_cols: 分区列 [('dt', 'STRING'), ...]
storage_format: 存储格式
location: HDFS路径
"""
# 列定义
col_defs = []
for col in columns:
col_name = col[0]
col_type = col[1]
col_comment = col[2] if len(col) > 2 else ''
if col_comment:
col_defs.append(f" `{col_name}` {col_type} COMMENT '{col_comment}'")
else:
col_defs.append(f" `{col_name}` {col_type}")
# 分区定义
part_defs = ""
if partition_cols:
part_strs = [f" `{p[0]}` {p[1]}" for p in partition_cols]
part_defs = f"PARTITIONED BY (\n" + ",\n".join(part_strs) + "\n)"
# 存储格式
if storage_format.upper() == 'PARQUET':
storage_clause = "STORED AS PARQUET"
elif storage_format.upper() == 'TEXTFILE':
storage_clause = "ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' STORED AS TEXTFILE"
elif storage_format.upper() == 'ORC':
storage_clause = "STORED AS ORC"
else:
storage_clause = f"STORED AS {storage_format}"
# HDFS路径
location_clause = f"LOCATION '{location}'" if location else ""
# 组合SQL
sql = f"""
CREATE EXTERNAL TABLE IF NOT EXISTS {table_name} (
{",\n".join(col_defs)}
)
{part_defs}
{storage_clause}
{location_clause}
TBLPROPERTIES (
'impala.table.statistics.enabled'='true',
'impala.enable.orc.schema.evolution'='false'
);
"""
return sql.strip()
# 使用示例
columns = [
('id', 'BIGINT', '用户ID'),
('name', 'STRING', '用户名'),
('age', 'INT', '年龄'),
('amount', 'DECIMAL(10,2)', '金额')
]
partition_cols = [('dt', 'STRING'), ('city', 'STRING')]
sql = generate_create_table_sql(
'db_name.user_info',
columns=columns,
partition_cols=partition_cols,
storage_format='PARQUET',
location='/user/hive/warehouse/db_name.db/user_info'
)
print(sql)
从配置文件生成建表语句
import json
import yaml
import configparser
class ImpalaTableConfigGenerator:
"""Impala表配置生成器"""
def __init__(self, config_file):
self.config = self._load_config(config_file)
def _load_config(self, config_file):
"""加载配置文件"""
if config_file.endswith('.json'):
with open(config_file, 'r') as f:
return json.load(f)
elif config_file.endswith('.yaml') or config_file.endswith('.yml'):
with open(config_file, 'r') as f:
return yaml.safe_load(f)
else:
raise ValueError("Unsupported config file format")
def generate_ddl(self, table_name=None):
"""生成DDL语句"""
table_config = self.config.get(table_name, self.config)
columns = []
for col in table_config['columns']:
col_tuple = (
col['name'],
col['type'],
col.get('comment', '')
)
columns.append(col_tuple)
partition_cols = [
(p['name'], p['type'])
for p in table_config.get('partition_columns', [])
]
return generate_create_table_sql(
table_name or table_config['table_name'],
columns,
partition_cols=partition_cols,
storage_format=table_config.get('storage_format', 'PARQUET'),
location=table_config.get('location', None)
)
def generate_all_ddls(self, output_file=None):
"""生成所有表的DDL"""
ddls = []
for table_name in self.config:
if isinstance(self.config[table_name], dict) and 'columns' in self.config[table_name]:
ddl = self.generate_ddl(table_name)
ddls.append(ddl)
if output_file:
with open(output_file, 'w') as f:
f.write('\n\n'.join(ddls))
return '\n\n'.join(ddls)
# YAML配置示例（table_config.yaml）
yaml_config = """
user_info:
columns:
- name: id
type: BIGINT
comment: 用户ID
- name: name
type: STRING
comment: 用户名
- name: age
type: INT
comment: 年龄
partition_columns:
- name: dt
type: STRING
storage_format: PARQUET
location: /user/hive/warehouse/test.db/user_info
order_info:
columns:
- name: order_id
type: STRING
comment: 订单ID
- name: user_id
type: BIGINT
comment: 用户ID
- name: amount
type: DECIMAL(10,2)
comment: 订单金额
- name: status
type: STRING
comment: 订单状态
partition_columns:
- name: dt
type: STRING
storage_format: PARQUET
"""
# 使用
with open('table_config.yaml', 'w') as f:
f.write(yaml_config)
generator = ImpalaTableConfigGenerator('table_config.yaml')
all_ddls = generator.generate_all_ddls('all_tables.sql')
print(all_ddls)
从数据库元数据生成
import pymysql
from sqlalchemy import create_engine, MetaData, Table
def generate_from_mysql(mysql_conn, database, table_name):
"""从MySQL表结构生成Impala建表语句"""
# 类型映射
type_mapping = {
'int': 'INT',
'bigint': 'BIGINT',
'varchar': 'STRING',
'char': 'STRING',
'text': 'STRING',
'decimal': 'DECIMAL',
'float': 'FLOAT',
'double': 'DOUBLE',
'datetime': 'TIMESTAMP',
'timestamp': 'TIMESTAMP',
'date': 'DATE',
'tinyint': 'TINYINT',
'smallint': 'SMALLINT',
}
# 连接到MySQL
conn = pymysql.connect(**mysql_conn)
cursor = conn.cursor()
# 获取表结构
cursor.execute(f"DESCRIBE {database}.{table_name}")
columns_info = cursor.fetchall()
# 获取注释
cursor.execute(f"""
SELECT COLUMN_NAME, COLUMN_COMMENT
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = '{database}' AND TABLE_NAME = '{table_name}'
""")
comments = dict(cursor.fetchall())
columns = []
for col_info in columns_info:
col_name = col_info[0]
mysql_type = col_info[1]
# 转换类型
impala_type = type_mapping.get(
mysql_type.split('(')[0].lower(),
'STRING'
)
# 保留精度
if 'decimal' in mysql_type.lower():
impala_type = mysql_type.upper()
comment = comments.get(col_name, '')
columns.append((col_name, impala_type, comment))
cursor.close()
conn.close()
return generate_create_table_sql(
f"{database}.{table_name}",
columns,
storage_format='PARQUET'
)
# 使用示例
mysql_config = {
'host': 'localhost',
'user': 'root',
'password': 'password',
'port': 3306
}
impala_ddl = generate_from_mysql(mysql_config, 'source_db', 'users')
print(impala_ddl)
从CSV文件生成
import csv
def generate_from_csv(csv_file_path, table_name, delimiter=','):
"""从CSV文件头生成Impala建表语句"""
# 读取CSV头部
with open(csv_file_path, 'r') as f:
reader = csv.reader(f, delimiter=delimiter)
headers = next(reader)
# 推断数据类型（简化版）
type_guesses = []
with open(csv_file_path, 'r') as f:
reader = csv.DictReader(f, delimiter=delimiter)
for row_num, row in enumerate(reader):
if row_num >= 100: # 只检查前100行
break
for i, header in enumerate(headers):
if len(type_guesses) <= i:
type_guesses.append({'INT': True, 'FLOAT': True, 'STRING': False})
value = row[header].strip()
if value:
# 尝试推断类型
try:
int(value)
except ValueError:
type_guesses[i]['INT'] = False
try:
float(value)
except ValueError:
type_guesses[i]['FLOAT'] = False
# 确定最终类型
columns = []
for i, header in enumerate(headers):
if type_guesses[i]['INT']:
col_type = 'INT'
elif type_guesses[i]['FLOAT']:
col_type = 'DECIMAL(20,10)'
else:
col_type = 'STRING'
columns.append((header, col_type, ''))
return generate_create_table_sql(table_name, columns, storage_format='PARQUET')
# 使用示例
csv_ddl = generate_from_csv('data.csv', 'csv_table')
print(csv_ddl)
生成表配置JSON
def generate_table_config(table_name, table_metadata):
"""生成表配置JSON"""
config = {
'table_name': table_name,
'database': table_metadata.get('database', 'default'),
'columns': [],
'partition_columns': [],
'properties': {
'format': table_metadata.get('format', 'PARQUET'),
'compression': table_metadata.get('compression', 'SNAPPY'),
'location': table_metadata.get('location', ''),
'row_format': table_metadata.get('row_format', ''),
'serde': table_metadata.get('serde', ''),
},
'statistics': {
'num_rows': table_metadata.get('num_rows', 0),
'total_size': table_metadata.get('total_size', 0),
},
'tblproperties': {
'impala.table.statistics.enabled': 'true',
'transient_lastDdlTime': str(int(time.time())),
}
}
for col in table_metadata.get('columns', []):
config['columns'].append({
'name': col['name'],
'type': col['type'],
'comment': col.get('comment', ''),
'nullable': col.get('nullable', True),
'default': col.get('default', None),
'statistics': {
'num_distinct': col.get('ndv', 0),
'num_nulls': col.get('num_nulls', 0),
}
})
for part in table_metadata.get('partition_columns', []):
config['partition_columns'].append({
'name': part['name'],
'type': part['type'],
'position': part.get('position', 0),
})
return config
# 使用示例
import time
test_metadata = {
'database': 'analytics',
'format': 'PARQUET',
'compression': 'SNAPPY',
'location': '/user/hive/warehouse/analytics.db/events',
'columns': [
{'name': 'event_id', 'type': 'STRING', 'comment': '事件ID', 'ndv': 1000000},
{'name': 'user_id', 'type': 'BIGINT', 'comment': '用户ID', 'ndv': 500000},
{'name': 'event_time', 'type': 'TIMESTAMP', 'comment': '事件时间'},
{'name': 'event_type', 'type': 'STRING', 'comment': '事件类型', 'ndv': 50},
],
'partition_columns': [
{'name': 'dt', 'type': 'STRING', 'position': 1},
{'name': 'hour', 'type': 'STRING', 'position': 2},
],
'num_rows': 10000000,
'total_size': 2147483648,
}
config = generate_table_config('events', test_metadata)
print(json.dumps(config, indent=2, ensure_ascii=False))
完整工具类
class ImpalaTableManager:
"""Impala表管理工具类"""
def __init__(self, impala_host='localhost', impala_port=21050):
self.host = impala_host
self.port = impala_port
def generate_ddl_from_spec(self, spec):
"""从表规范生成DDL"""
return generate_create_table_sql(
f"{spec['database']}.{spec['table_name']}",
spec['columns'],
partition_cols=spec.get('partition_columns', []),
storage_format=spec.get('storage_format', 'PARQUET'),
location=spec.get('location', '')
)
def execute_ddl(self, ddl):
"""执行DDL语句"""
from impala.dbapi import connect
conn = connect(host=self.host, port=self.port)
cursor = conn.cursor()
try:
cursor.execute(ddl)
conn.commit()
print(f"Successfully executed DDL")
except Exception as e:
print(f"Error executing DDL: {e}")
conn.rollback()
finally:
cursor.close()
conn.close()
def save_config(self, table_name, config, output_file):
"""保存表配置"""
if output_file.endswith('.json'):
with open(output_file, 'w') as f:
json.dump(config, f, indent=2, ensure_ascii=False)
elif output_file.endswith('.yaml'):
with open(output_file, 'w') as f:
yaml.dump(config, f, default_flow_style=False)
# 完整使用示例
manager = ImpalaTableManager()
# 生成DDL
spec = {
'database': 'test_db',
'table_name': 'user_behavior',
'columns': [
('user_id', 'BIGINT', '用户ID'),
('behavior', 'STRING', '行为类型'),
],
'partition_columns': [('dt', 'STRING')],
'storage_format': 'PARQUET'
}
ddl = manager.generate_ddl_from_spec(spec)
print("Generated DDL:")
print(ddl)
这些方法可以根据你的具体需求灵活调整,支持从各种数据源生成Impala表配置和DDL语句。
