使用 S3ToRedshiftOperator 截断红移表时出错

Question

我想在将 CSV 文件加载到 Redshift 表之前截断我的 Redshift 表。

错误： airflow.exceptions.AirflowException：无效的参数被传递给 S3ToRedshiftOperator (task_id: dag_run_s3_to_redshift)。无效的参数是： **kwargs: {'method': 'REPLACE'}

以下代码：

task_fail_s3_to_redshift =  S3ToRedshiftOperator(
        s3_bucket=S3_BUCKET,
        s3_key="{{ti.xcom_pull(task_ids='export_db',key='FILE_PATH_1')}}",
        schema="dw_stage",
        table="task_fail",
        copy_options=['csv',"IGNOREHEADER 1"],
        redshift_conn_id='redshift',
        method='REPLACE',
        task_id='task_fail_s3_to_redshift',
    ) 

 start >> task_fail_s3_to_redshift >> end 

Answer 1

method 参数已添加到 PR 中，适用于：

apache-airflow-providers-amazon >= 2.4.0

您遇到的错误意味着您使用的是旧版本的亚马逊提供商，这就是它不适合您的原因。

您的选择是：

1.升级提供者

pip install apache-airflow-providers-amazon --upgrade

2.如果升级不是一个选项，则使用已弃用的truncate_table 参数：

task_fail_s3_to_redshift =  S3ToRedshiftOperator(
        s3_bucket=S3_BUCKET,
        s3_key="{{ti.xcom_pull(task_ids='export_db',key='FILE_PATH_1')}}",
        schema="dw_stage",
        table="task_fail",
        copy_options=['csv',"IGNOREHEADER 1"],
        redshift_conn_id='redshift',
        truncate_table=True,
        task_id='task_fail_s3_to_redshift',
    ) 

由于您想要截断选项 - 它会为您提供相同的功能。