Airflow SFTPToS3Operator - download multiple files from STFP to S3

user_17754428 0
Viewed 348

I am new to Airflow. My requirement is to develop a data pipeline to do below:

i) Connect and download multiple csv files (filename starting with name) airflow from given SFTP folder to S3

ii) Ingest files from S3 to redshift

I have used SFTPToS3Operator and S3ToRedshiftOperator to do this and it's working perfectly fine for one file when I provide full file name for sftp_path and s3_key, but it's failing with IO error if I provide wild card (airflow*.csv) for file name in sftp_path and s3_key.

Need your expert opinion on approach to download multiple files from SFTP to s3 and ingesting into redshift

I am using below code for single file, which is working fine for single file

 # Dag initialization with required parameter
     with DAG("sftp_to_redshift_data_ingestion", start_date=datetime(2021, 1 ,1), 
      schedule_interval="@daily", default_args=default_args, catchup=False) as dag:

  # SFTP to S3 data transfer
    sftp_to_s3 = SFTPToS3Operator(
        task_id="sftp_to_s3",
        sftp_conn_id="sftp_conn",
        sftp_path="/Inbound/airflow_demo.csv"
        s3_conn_id="s3_dev_conn",
        s3_bucket="s3_dev_bucket",
        s3_key="input/test/airflow_demo.csv" 
    )

    #S3 to Redshift data transfer
    transfer_s3_to_redshift = S3ToRedshiftOperator(task_id='transfer_s3_to_redshift',
        aws_conn_id = 's3_dev_conn',
        redshift_conn_id = "redshift_dev_conn",
        s3_bucket="s3_dev_bucket",
        s3_key="input/test/airflow_demo.csv",
        schema="test",
        table="airflow_demo_table",
        column_list=["customer_id", "customer_name", "customer_address"],
        copy_options= ["csv", "DELIMITER ','", "FILLRECORD", "IGNOREHEADER 1", "COMPROWS 1000000"]
        )
airflow-2.x airflow
1 Answers

The SFTPToS3Operator only copies over one file at a time. You would need to first get a list of all the file names (metadata) from SFTP. Loop through the files and run the SFTPtoS3 operator, it will copy all the files into S3.

EX:

files = ["file1.csv", "file2.csv"]
for file in files:
   sftp_to_s3 = SFTPToS3Operator(
        task_id="sftp_to_s3",
        sftp_conn_id="sftp_conn",
        sftp_path=f"/Inbound/{file}"
        s3_conn_id="s3_dev_conn",
        s3_bucket="s3_dev_bucket",
        s3_key=f"input/test/{file}" 
    )
user_118614940
Related