Mailing-List: contact commits-help@airflow.incubator.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@airflow.incubator.apache.org
Date: Thu, 28 Apr 2016 21:14:12 +0000 (UTC)
From: "Chris Riccomini (JIRA)" <jira@apache.org>
To: commits@airflow.incubator.apache.org
Message-ID: <JIRA.12963660.1461877680000.64168.1461878052927@Atlassian.JIRA>
In-Reply-To: <JIRA.12963660.1461877680000@Atlassian.JIRA>
References: <JIRA.12963660.1461877680000@Atlassian.JIRA>
 <JIRA.12963660.1461877680755@arcas>
Subject: [jira] [Commented] (AIRFLOW-19) How can I have an Operator B
 iterate over a list returned from upstream by Operator A?
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit


    [ https://issues.apache.org/jira/browse/AIRFLOW-19?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15263013#comment-15263013 ] 

Chris Riccomini commented on AIRFLOW-19:
----------------------------------------

Code from gist is:

{code}
t1 = DistCpSensor(
    task_id='sensor',
    lookback=2,
    distcp_dao=distcp_dao,
    source_conn=source_conn,
    parent_paths=['/tmp/testAirflowDistCp'],
    timeout=12*60*60,
    dag=dag)

def run_distcp_on_each(*args, **kwargs):
    ti = kwargs['ti']
    pprint(kwargs)
    to_process = ti.xcom_pull(task_ids=None, key='to_process')
    pprint(to_process)
    for work in to_process:
        t3 = DistCpOperator(
            task_id='distcp_command', 
            source_conn=source_conn,
            work=work,
            dag=dag)

        t4 = BashOperator(
            task_id='run_distcp',
            bash_command="{{ ti.xcom_pull(task_ids='distcp_command') }}",
            xcom_push=True,
            env=os.environ.copy(),
            dag=dag)

        t5 = BashOperator(
            task_id='get_application_id',
            bash_command="echo {{ ti.xcom_pull(task_ids='run_distcp') }} | awk '{print $NF}' | sed 's/job/application/g'",
            xcom_push=True,
            env=os.environ.copy(),
            dag=dag)

        t6 = DistCpMonitor(
            task_id='monitor',
            application_id="{{ ti.xcom_pull(task_ids='get_application_id') }}",
            resource_manager_conn=resource_manager_conn,
            dag=dag)

        t7 = DistCpSensorCompletionOperator(
            task_id='mysql_update',
            distcp_dao=distcp_dao,
            dag=dag)

        t3.set_upstream(t2)
        t4.set_upstream(t3)
        t5.set_upstream(t4)
        t6.set_upstream(t5)
        t7.set_upstream(t6)

t2 = PythonOperator(
    task_id='run_distcp_on_each',
    provide_context=True,
    python_callable=run_distcp_on_each,
    dag=dag)

t2.set_upstream(t1)
{code}

> How can I have an Operator B iterate over a list returned from upstream by Operator A?
> --------------------------------------------------------------------------------------
>
>                 Key: AIRFLOW-19
>                 URL: https://issues.apache.org/jira/browse/AIRFLOW-19
>             Project: Apache Airflow
>          Issue Type: Bug
>            Reporter: Praveenkumar Venkatesan
>            Priority: Minor
>              Labels: support
>
> Here is what I am trying to do exactly: https://gist.github.com/praveev/7b93b50746f8e965f7139ecba028490a
> the python operator log just returns the following
> [2016-04-28 11:56:22,296] {models.py:1041} INFO - Executing <Task(PythonOperator): run_distcp_on_each> on 2016-04-28 11:56:12
> [2016-04-28 11:56:22,350] {python_operator.py:66} INFO - Done. Returned value was: None
> it didn't even print my kwargs and to_process data
> To simplify this. Lets say t1 returns 3 elements. I want to iterate over the list and run t2 -> t3 for each element.


--
This message was sent by Atlassian JIRA
(v6.3.4#6332)