Airflow Correct way for passing _XcomLazyAccess values to another task

hulike2286 0
Viewed 28

I am trying to pass values which are returned from a task after using expand. The return type if you use dynamic mapping and expand in Airflow is the return type changes to _LazyXComAccess which is essentially a list.

    t1 = PythonOperator.partial(
        task_id="invoke_lambda",
        retries=1,
        retry_delay=timedelta(seconds=30),
        python_callable=invoke_lambda_function,
    ).expand(op_kwargs=generate_lambda_config())

How do you access the values and use it to trigger a following task for each value from the returned task?

I have tried using it directly like the following

t2 = invoke_second_lambda.expand(payload=XComArg(t1))

However this throws

botocore.errorfactory.TooManyRequestsException: An error occurred 
(TooManyRequestsException) when calling the Invoke operation (reached 
max retries: 4): Rate Exceeded.

Any ideas?

airflow python
1 Answers

When you create the first task:

t1 = PythonOperator.partial(
        task_id="invoke_lambda",
        retries=1,
        retry_delay=timedelta(seconds=30),
        python_callable=invoke_lambda_function,
    ).expand(op_kwargs=generate_lambda_config())

t1 will be the _LazyXComAccess which is XCom list calculated at runtime, so you cannot access the values before that. If you want to trigger a second lambda for each value returned from the first lambda, you can do exactly like you did in the first task:

t2 = PythonOperator.partial(
        task_id="invoke_second_lambda",
        retries=1,
        retry_delay=timedelta(seconds=30),
        python_callable=invoke_second_lambda,
    ).expand(op_kwargs=t1)

The expand method will load the XCom from the first task at runtime, and expand them to create a task for each element.

If the task t1 return an int for example, and you want to provide a payload {"my_int_input": <int from t1>}, you can create an intermediate task to transform the t1 output, and use its output as input for t1:

t2_input = PythonOperator.partial(
        task_id="transform_t1_output",
        python_callable=lambda: [{"my_int_input": val} for val in t1],
    )
Hussein Awala0
Related