使用S3ToRedShitOperator执行截断红移表时出错

问题描述

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

错误： airflow.exceptions.AirflowException：传递给S3ToRedshitOperator(task_id：dag_run_s3_to_redshift)的参数无效。无效参数为： **kwargs：{‘method’：‘place’}

以下代码：

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 

---

解决方案

method参数添加在PR中：

apache-airflow-providers-amazon >= 2.4.0

您遇到的错误意味着您使用的是旧版本的Amazon Provider，这就是它不适用于您的原因。

您的选项包括：

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',
    ) 

因为您需要截断选项-它将为您提供相同的功能。