Python脚本如何生成Impala表配置

wen 实用脚本 2026-07-05 169

本文目录导读：

生成Impala建表语句
从配置文件生成建表语句
从数据库元数据生成
从CSV文件生成
生成表配置JSON
完整工具类

我来详细说明如何用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语句。

本文地址： https://www.bmjup.com/post/16178.html