Mailing-List: contact dev-help@airflow.incubator.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@airflow.incubator.apache.org
MIME-Version: 1.0
From: Aaron Polhamus <aaron@credijusto.com>
Date: Tue, 25 Jul 2017 15:58:01 -0700
Message-ID: <CACP1=KmivXS3fb09d3rBxcM9enhOYMzaWyM0AGXsozRx1vGyBA@mail.gmail.com>
Subject: How to make DAGs dynamic based on task output
To: dev@airflow.incubator.apache.org
Content-Type: multipart/alternative; boundary="001a11457ec6701d0d05552c43a3"
archived-at: Tue, 25 Jul 2017 22:58:07 -0000

--001a11457ec6701d0d05552c43a3
Content-Type: text/plain; charset="UTF-8"
Content-Transfer-Encoding: quoted-printable

This question was posted on StackOverflow here:
https://stackoverflow.com/questions/45314174/how-to-dynamically-iterate-ove=
r-the-output-of-an-upstream-task-to-create-paralle

Consider the following example of a DAG where the first task, get_id_creds,
extracts a list of credentials from a database. This operation tells me
what users in my database I am able to run further data preprocessing on
and it writes those ids to the file /tmp/ids.txt. I then scan those ids
into my DAG and use them to generate a list of upload_transaction tasks
that can be run in parallel.

My question is: Is there a more idiomatically correct, dynamic way to do
this using airflow? What I have here feels clumsy and brittle. How can I
directly pass a list of valid IDs from one process to that defines the
subsequent downstream processes?

from datetime import datetime, timedeltaimport osimport sys
from airflow.models import DAGfrom airflow.operators.python_operator
import PythonOperator
import ds_dependencies

SCRIPT_PATH =3D os.getenv('DASH_PREPROC_PATH')if SCRIPT_PATH:
    sys.path.insert(0, SCRIPT_PATH)
    import dash_workerselse:
    print('Define DASH_PREPROC_PATH value in environmental variables')
    sys.exit(1)

default_args =3D {
  'start_date': datetime.now(),
  'schedule_interval': None}

DAG =3D DAG(
  dag_id=3D'dash_preproc',
  default_args=3Ddefault_args)

get_id_creds =3D PythonOperator(
    task_id=3D'get_id_creds',
    python_callable=3Ddash_workers.get_id_creds,
    provide_context=3DTrue,
    dag=3DDAG)
with open('/tmp/ids.txt', 'r') as infile:
    ids =3D infile.read().splitlines()
for uid in uids:
    upload_transactions =3D PythonOperator(
        task_id=3Duid,
        python_callable=3Ddash_workers.upload_transactions,
        op_args=3D[uid],
        dag=3DDAG)
    upload_transactions.set_downstream(get_id_creds)

Thanks in advance!
--=20


*Aaron Polhamus*
*Director of Data Science *

Cel (M=C3=A9xico): +52 (55) 1951-5612
Cell (USA): +1 (206) 380-3948
Tel: +52 (55) 1168 9757 - Ext. 181

--=20
***Por favor referirse a nuestra p=C3=A1gina web=20
<https://www.credijusto.com/aviso-de-privacidad/> para m=C3=A1s informaci=
=C3=B3n=20
acerca de nuestras pol=C3=ADticas de privacidad.*


--001a11457ec6701d0d05552c43a3--