Airflow s3 connection using UI

by

EDIT: This answer stores your secret key in plain text which can be a security risk and is not recommended. The best way is to put access key and secret key in the login/password fields, as mentioned in other answers below.
END EDIT

It’s hard to find references, but after digging a bit I was able to make it work.

TLDR

Create a new connection with the following attributes:

Conn Id: my_conn_S3

Conn Type: S3

Extra:

{"aws_access_key_id":"_your_aws_access_key_id_", "aws_secret_access_key": "_your_aws_secret_access_key_"}

Long version, setting up UI connection:

  • On Airflow UI, go to Admin > Connections
  • Create a new connection with the following attributes:
  • Conn Id: my_conn_S3
  • Conn Type: S3
  • Extra: {"aws_access_key_id":"_your_aws_access_key_id_", "aws_secret_access_key": "_your_aws_secret_access_key_"}
  • Leave all the other fields (Host, Schema, Login) blank.

To use this connection, below you can find a simple S3 Sensor Test. The idea of this test is to set up a sensor that watches files in S3 (T1 task) and once below condition is satisfied it triggers a bash command (T2 task).

Testing

  • Before running the DAG, ensure you’ve an S3 bucket named ‘S3-Bucket-To-Watch’.
  • Add below s3_dag_test.py to airflow dags folder (~/airflow/dags)
  • Start airflow webserver.
  • Go to Airflow UI (http://localhost:8383/)
  • Start airflow scheduler.
  • Turn on ‘s3_dag_test’ DAG on the main DAGs view.
  • Select ‘s3_dag_test’ to show the dag details.
  • On the Graph View you should be able to see it’s current state.
  • ‘check_s3_for_file_in_s3’ task should be active and running.
  • Now, add a file named ‘file-to-watch-1’ to your ‘S3-Bucket-To-Watch’.
  • First tasks should have been completed, second should be started and finish.

The schedule_interval in the dag definition is set to ‘@once’, to facilitate debugging.

To run it again, leave everything as it’s, remove files in the bucket and try again by selecting the first task (in the graph view) and selecting ‘Clear’ all ‘Past’,’Future’,’Upstream’,’Downstream’ …. activity. This should kick off the DAG again.

Let me know how it went.

s3_dag_test.py ;

"""
S3 Sensor Connection Test
"""

from airflow import DAG
from airflow.operators import SimpleHttpOperator, HttpSensor,   BashOperator, EmailOperator, S3KeySensor
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2016, 11, 1),
    'email': ['[email protected]'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 5,
    'retry_delay': timedelta(minutes=5)
}

dag = DAG('s3_dag_test', default_args=default_args, schedule_interval="@once")

t1 = BashOperator(
    task_id='bash_test',
    bash_command='echo "hello, it should work" > s3_conn_test.txt',
    dag=dag)

sensor = S3KeySensor(
    task_id='check_s3_for_file_in_s3',
    bucket_key='file-to-watch-*',
    wildcard_match=True,
    bucket_name="S3-Bucket-To-Watch",
    s3_conn_id='my_conn_S3',
    timeout=18*60*60,
    poke_interval=120,
    dag=dag)

t1.set_upstream(sensor)

Main References:

More Related Contents:

  1. Pip error even Microsoft Visual C++ 14.0 is installed
  2. How can I reconnect to the browser opened by webdriver with selenium?
  3. UnicodeEncodeError: ‘charmap’ codec can’t encode character ‘\u2010’: character maps to [duplicate]
  4. how to set proxy with authentication in selenium chromedriver python?
  5. Neural network always predicts the same class
  6. Since matplotlib.finance has been deprecated, how can I use the new mpl_finance module?
  7. Writing a dictionary to a text file?
  8. Removing black background and make transparent from grabcut output in python open cv
  9. Passing Variables To Google Cloud Functions
  10. urllib.error.URLError:
  11. Sending selenium chrome instance to the background using Python
  12. selenium.common.exceptions.SessionNotCreatedException: Message: session not created from tab crashed using ChromeDriver Chrome Selenium Python
  13. Python async: Waiting for stdin input while doing other stuff
  14. How do I remove/delete a virtualenv?
  15. ‘Import “Path.to.own.script” could not be resolved Pylance (reportMissingImports)’ in VS Code using Python 3.x on Ubuntu 20.04 LTS
  16. Tkinter module not found on Ubuntu
  17. How do I create a variable-length input LSTM in Keras?
  18. vcvarsall.bat needed for python to compile missing from visual studio 2015 ( v 14)
  19. in VS Code ImportError: cannot import name ‘Mapping’ from ‘collections’
  20. Flexible appending new data to yaml files
  21. How i can get new ip from tor every requests in threads?
  22. ImportError: libcublas.so.9.0: cannot open shared object file
  23. How to send real-time sensor data to PC from Raspberry Pi Zero?
  24. Discord Music bot VoiceClient’ object has no attribute ‘create_ytdl_player’
  25. How to bypass being rate limited ..HTML Error 1015 using Python
  26. How do I print colored output with Python 3?
  27. selenium.common.exceptions.WebDriverException: Message: unknown error: unable to discover open pages using ChromeDriver through Selenium
  28. Saving StandardScaler() model for use on new datasets
  29. Debugging Numpy VisibleDeprecationWarning (ndarray from ragged nested sequences)
  30. How to determine what version of python3 tkinter is installed on my linux machine?

Leave a Comment