如何不使用S3桶通过AirFlow实现RedShift同一库不同Schema的数据拷贝

无S3依赖的实现原理 #

RedShift同一数据库下的不同Schema属于同一个集群实例，不需要跨实例/跨库的外部中转，直接用原生SQL即可完成数据拷贝，全程不涉及S3存储桶操作。你只需要确保执行SQL的数据库账号拥有：

• 源Schema的SELECT权限
• 目标Schema的CREATE、INSERT权限

常用拷贝场景SQL示例 #

全量覆盖同步 #

如果需要每次同步都覆盖目标表的全部数据，操作如下：

1. 首次创建目标表时保留源表所有属性（分布键、约束、索引、默认值等）

CREATE TABLE IF NOT EXISTS target_schema.target_table 
(LIKE source_schema.source_table INCLUDING DEFAULTS INCLUDING CONSTRAINTS INCLUDING INDEXES);

2. 清空目标表存量后插入全量数据

TRUNCATE TABLE target_schema.target_table;
INSERT INTO target_schema.target_table SELECT * FROM source_schema.source_table;

如果不需要保留原表属性，也可以直接用CTAS语句一步完成：
CREATE TABLE IF NOT EXISTS target_schema.target_table AS SELECT * FROM source_schema.source_table;

增量同步 #

如果只需要同步新增/变更的数据，添加过滤条件即可，例如按日期字段同步前一天的数据：

INSERT INTO target_schema.target_table 
SELECT * FROM source_schema.source_table 
WHERE dt = '{{ prev_ds }}'; -- 支持AirFlow模板变量动态传时间

AWS 托管AirFlow（MWAA）DAG示例 #

前置要求 #

• MWAA环境已安装apache-airflow-providers-amazon依赖（官方默认镜像已自带）
• AirFlow WebUI的Connections页面已配置RedShift连接，连接ID为redshift_default，连接类型选择Amazon Redshift，填写对应集群的地址、端口、数据库名、认证信息

DAG代码 #

from datetime import datetime, timedelta
from airflow import DAG
from airflow.providers.amazon.aws.operators.redshift_sql import RedshiftSQLOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    dag_id='redshift_cross_schema_data_copy',
    default_args=default_args,
    description='RedShift同库跨Schema数据同步，无S3依赖',
    schedule_interval=timedelta(days=1),
    catchup=False,
    tags=['redshift', 'data_sync', 'cross_schema'],
) as dag:

    # 任务1：创建目标表（不存在时）
    create_target_table = RedshiftSQLOperator(
        task_id='create_target_table',
        redshift_conn_id='redshift_default',
        sql="""
        CREATE TABLE IF NOT EXISTS target_schema.target_table 
        (LIKE source_schema.source_table INCLUDING DEFAULTS INCLUDING CONSTRAINTS INCLUDING INDEXES);
        """
    )

    # 任务2：清空目标表（全量同步场景保留，增量同步请删除此任务）
    truncate_target = RedshiftSQLOperator(
        task_id='truncate_target_table',
        redshift_conn_id='redshift_default',
        sql="TRUNCATE TABLE target_schema.target_table;"
    )

    # 任务3：执行跨Schema数据拷贝
    sync_data = RedshiftSQLOperator(
        task_id='sync_cross_schema_data',
        redshift_conn_id='redshift_default',
        sql="""
        INSERT INTO target_schema.target_table
        SELECT * FROM source_schema.source_table;
        -- 增量同步替换为带过滤条件的SQL即可，例如：
        -- SELECT * FROM source_schema.source_table WHERE dt = '{{ prev_ds }}';
        """
    )

    # 任务依赖配置
    create_target_table >> truncate_target >> sync_data

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