Connection is actually a model which you can use to query and insert a new connection from airflow import settings from airflow.models import Connection conn = Connection( conn_id=conn_id, conn_type=conn_type, host=host, login=login, password=password, port=port ) #create a connection object session = settings.Session() # get the session session.add(conn) session.commit() # it will insert the connection object programmatically.
NOT available for airflow 2.x. I think that I just figured it out: Create a SSH connection in UI under Admin > Connection. Note: the connection will be deleted if you reset the database In the Python file add the following from airflow.contrib.hooks import SSHHook sshHook = SSHHook(conn_id=<YOUR CONNECTION ID FROM THE UI>) Add the … Read more
I use something like this. A project is normally something completely separate or unique. Perhaps DAGs to process files that we receive from a certain client which will be completely unrelated to everything else (almost certainly a separate database schema) I have my operators, hooks, and some helper scripts (delete all Airflow data for a … Read more
Setting up SMTP Server for Airflow Email alerts using Gmail: Create an email id from which you want to send alerts about DAG failure or if you want to use EmailOperator. Edit airflow.cfg file to edit the smtp details for the mail server. For demo you can use any gmail account. Create a google App … Read more
Just hit this myself. its an issue with SQLAlchemy dependency to fix I did the following: pip3 uninstall SQLAlchemy pip3 install SQLAlchemy==1.3.15 https://github.com/apache/airflow/issues/8211
For a great overview of Airflow and Apache NiFi checkout this reddit post: https://www.reddit.com/r/bigdata/comments/51mgk6/comparing_airbnb_airflow_and_apache_nifi/ For your specific use-case of ingesting Json files, enriching them and routing them to Kafka I believe NiFi is the right tool for the job. A couple of processors you could potentially use, as well as documentation for each, are below: … Read more
You will need to use LocalExecutor. Check your configs (airflow.cfg), you might be using SequentialExectuor which executes tasks serially. Airflow uses a Backend database to store metadata. Check your airflow.cfg file and look for executor keyword. By default, Airflow uses SequentialExecutor which would execute task sequentially no matter what. So to allow Airflow to run … Read more
All operators have an argument trigger_rule which can be set to ‘all_done’, which will trigger that task regardless of the failure or success of the previous task(s). You could set the trigger rule for the task you want to run to ‘all_done’ instead of the default ‘all_success’. A simple bash operator task with that argument … Read more
Run airflow dags list (or airflow list_dags for Airflow 1.x) to check, whether the dag file is located correctly. For some reason, I didn’t see my dag in the browser UI before I executed this. Must be issue with browser cache or something. If that doesn’t work, you should just restart the webserver with airflow … Read more
Pass a dict object to op_kwargs Use the keys to access their value from kwargs dict in your python callable def SendEmail(**kwargs): print(kwargs[‘key1’]) print(kwargs[‘key2’]) msg = MIMEText(“The pipeline for client1 is completed, please check.”) msg[‘Subject’] = “xxxx” msg[‘From’] = “xxxx” …… s = smtplib.SMTP(‘localhost’) s.send_message(msg) s.quit() t5_send_notification = PythonOperator( task_id=’t5_send_notification’, provide_context=True, python_callable=SendEmail, op_kwargs={‘key1’: ‘value1’, ‘key2’: … Read more