Apache Airflow S3ListOperator not listing files

user_15778874 0
Viewed 2220

I am trying to use the airflow.providers.amazon.aws.operators.s3_list S3ListOperator to list files in an S3 bucket in my AWS account with the DAG operator below:

list_bucket = S3ListOperator(
    task_id = 'list_files_in_bucket',
    bucket = '<MY_BUCKET>',
    aws_conn_id = 's3_default'
)

I have configured my Extra Connection details in the form of: {"aws_access_key_id": "<MY_ACCESS_KEY>", "aws_secret_access_key": "<MY_SECRET_KEY>"}

When I run my Airflow job, it appears it is executing fine & my task status is Success. Here is the Log output:

[2021-04-27 11:44:50,009] {base_aws.py:368} INFO - Airflow Connection: aws_conn_id=s3_default
[2021-04-27 11:44:50,013] {base_aws.py:170} INFO - Credentials retrieved from extra_config
[2021-04-27 11:44:50,013] {base_aws.py:84} INFO - Creating session with aws_access_key_id=<MY_ACCESS_KEY> region_name=None
[2021-04-27 11:44:50,027] {base_aws.py:157} INFO - role_arn is None
[2021-04-27 11:44:50,661] {taskinstance.py:1185} INFO - Marking task as SUCCESS. dag_id=two_step, task_id=list_files_in_bucket, execution_date=20210427T184422, start_date=20210427T184439, end_date=20210427T184450
[2021-04-27 11:44:50,676] {taskinstance.py:1246} INFO - 0 downstream tasks scheduled from follow-on schedule check
[2021-04-27 11:44:50,700] {local_task_job.py:146} INFO - Task exited with return code 0

Is there anything I can do to print the files in my bucket to Logs? TIA

airflow amazon-web-services amazon-s3
2 Answers

This code is enough and you don't need to use print function. Just check the corresponding log, then go to xcom, and the return list is there.

list_bucket = S3ListOperator(
        task_id='list_files_in_bucket',
        bucket='ob-air-pre',
        prefix='data/',
        delimiter='/',
        aws_conn_id='aws'
    )

enter image description here

user_41698920

The result from executing S3ListOperator is an XCom object that is stored in the Airflow database after the task instance has completed.

You need to declare another operator to feed in the results from the S3ListOperator and print them out.

For example in Airflow 2.0.0 and up you can use TaskFlow:

from airflow.models import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.utils import timezone

dag = DAG(
    dag_id='my-workflow',
    start_date=timezone.parse('2021-01-14 21:00')
)

@dag.task(task_id="print_objects")
def print_objects(objects):
    print(objects)


list_bucket = S3ListOperator(
    task_id='list_files_in_bucket',
    bucket='<MY_BUCKET>',
    aws_conn_id='s3_default',
    dag=dag
)


print_objects(list_bucket.output)

In older versions,

from airflow.models import DAG
from airflow.operators.python import PythonOperator
from airflow.utils import timezone

dag = DAG(
    dag_id='my-workflow',
    start_date=timezone.parse('2021-01-14 21:00')
)
def print_objects(objects):
    print(objects)


list_bucket = S3ListOperator(
    dag=dag,
    task_id='list_files_in_bucket',
    bucket='<MY_BUCKET>',
    aws_conn_id='s3_default',
)

print_objects_in_bucket = PythonOperator(
     dag=dag,
     task_id='print_objects_in_bucket',
     python_callable=print_objects,
     op_args=("{{ti.xcom_pull(task_ids='list_files_in_bucket')}}",)
)

list_bucket >> print_objects_in_bucket
user_51898110
Related