如何向分区SQL表插入DataFrame数据：分步教程

import pandas as pd
from sqlalchemy import create_engine
from pyhive import hive # 假设目标数据库是Hive

# 示例DataFrame
data = {'col1': [1, 2, 3], 'col2': ['A', 'B', 'C'], 'dt_partition': ['2024-03-26', '2024-03-26', '2024-03-27']}
df = pd.DataFrame(data)

# 配置Hive的SQLAlchemy引擎
# 请根据实际环境修改host, port, database, username等
hive_engine = create_engine(
    'hive://your_username@localhost:10000/your_database',
    connect_args={'auth': 'NOSASL'} # 或其他认证方式
)

# 定义临时表名称
temp_table_name = 'my_table_tmp'

# 将DataFrame数据写入临时表
# if_exists='replace' 会在每次执行时替换临时表，确保数据干净
# index=False 避免将DataFrame的索引作为一列写入数据库
df.to_sql(temp_table_name, hive_engine, if_exists='replace', index=False, method='multi')

print(f"数据已成功写入临时表：{temp_table_name}")

# 假设目标分区表名为 'my_partitioned_table'
# 假设分区列为 'dt' (日期), 格式为 YYYYMMDD
target_table_name = 'my_partitioned_table'
partition_column = 'dt'
partition_value = '20240326' # 示例：插入到2024年3月26日的分区

# 建立PyHive连接
# 这与SQLAlchemy引擎是独立的，用于执行原生SQL
hive_conn = hive.connect(host='localhost',
                         port=10000,
                         username='your_username',
                         database='your_database')

try:
    with hive_conn.cursor() as cursor:
        # 构建INSERT OVERWRITE TABLE语句
        # 注意：INSERT OVERWRITE TABLE会覆盖指定分区中所有现有数据
        # 如果需要追加数据到分区，应使用 INSERT INTO TABLE ... PARTITION(...) SELECT ...
        insert_sql = f"""
        INSERT OVERWRITE TABLE {target_table_name} PARTITION({partition_column}='{partition_value}')
        SELECT col1, col2 FROM {temp_table_name}
        WHERE dt_partition = '{partition_value[:4]}-{partition_value[4:6]}-{partition_value[6:]}'
        """
        # 注意：SELECT的列名应与目标表列名匹配
        # WHERE子句用于筛选出属于当前分区的数据，这在临时表可能包含多个分区数据时非常重要

        cursor.execute(insert_sql)
        print(f"数据已成功从临时表 {temp_table_name} 插入到分区表 {target_table_name} 的分区 {partition_column}={partition_value}")
    hive_conn.commit() # 提交事务
except Exception as e:
    hive_conn.rollback() # 发生错误时回滚
    print(f"数据插入失败: {e}")
finally:
    hive_conn.close() # 关闭连接