Presto是否支持Insert Overwrite写入Hive分区表？求正确语法

1. 核心INSERT OVERWRITE语法 #

针对分区表，分两种场景使用：

• 覆盖指定单个/多个分区：

  INSERT OVERWRITE TABLE your_schema.your_table PARTITION (partition_col='target_value')
  SELECT col1, col2, ... FROM source_table WHERE condition;
  

  注：SELECT语句只需包含目标表中非分区列的字段，数量、类型必须与目标表严格匹配。
• 覆盖全表所有分区：

  INSERT OVERWRITE TABLE your_schema.your_table
  SELECT col1, col2, partition_col FROM source_table;
  

  注：此时必须将分区列包含在SELECT结果中，且字段顺序要与目标表结构完全对应。

2. 常见报错排查与解决 #

• 权限不足：确保Airflow执行账号拥有目标表的分区删除、写入权限（INSERT OVERWRITE本质是先删除对应分区再插入数据）。
• 分区列格式不匹配：分区列的类型、格式必须与表定义一致，比如表分区列是DATE类型，就不能用'2024/05/20'这种格式，必须写'2024-05-20'。
• 动态分区未开启：如果使用动态分区（即SELECT结果自动匹配分区），需确保Trino对应Catalog（如Hive）开启了相关配置，可在会话级别临时设置：

  SET SESSION hive.insert-existing-partitions-behavior = 'OVERWRITE';
  

3. 备选方案：先删分区再插入 #

如果INSERT OVERWRITE仍报错，可拆分两步操作：

• 删除指定分区：

  ALTER TABLE your_schema.your_table DROP PARTITION IF EXISTS (partition_col='target_value');
  

  加IF EXISTS可避免分区不存在时触发报错。
• 插入数据到分区：

  INSERT INTO your_schema.your_table PARTITION (partition_col='target_value')
  SELECT col1, col2 FROM source_table WHERE condition;
  

4. Airflow PrestoSqlOperator示例 #

from airflow import DAG
from airflow.providers.presto.operators.presto import PrestoSqlOperator
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 5, 20),
}

with DAG('trino_partition_operation', default_args=default_args, schedule_interval='@daily') as dag:
    # 方案1：直接覆盖指定分区
    overwrite_task = PrestoSqlOperator(
        task_id='overwrite_target_partition',
        presto_conn_id='your_presto_connection',
        sql="""
            INSERT OVERWRITE TABLE your_schema.target_table PARTITION (dt='{{ ds }}')
            SELECT id, username FROM your_schema.source_table WHERE dt='{{ ds }}';
        """
    )

    # 方案2：先删分区再插入
    drop_partition = PrestoSqlOperator(
        task_id='drop_existing_partition',
        presto_conn_id='your_presto_connection',
        sql="ALTER TABLE your_schema.target_table DROP PARTITION IF EXISTS (dt='{{ ds }}');"
    )

    insert_data = PrestoSqlOperator(
        task_id='insert_into_partition',
        presto_conn_id='your_presto_connection',
        sql="""
            INSERT INTO your_schema.target_table PARTITION (dt='{{ ds }}')
            SELECT id, username FROM your_schema.source_table WHERE dt='{{ ds }}';
        """
    )

    drop_partition >> insert_data

内容的提问来源于stack exchange，提问作者Deepak