CODEWITHSUNDEEP
airflow
rescdsk
rescdsk

Reputation: 8895

In Airflow, what do they want you to do instead of pass data between tasks

In the docs, they say that you should avoid passing data between tasks:

This is a subtle but very important point: in general, if two operators need to share information, like a filename or small amount of data, you should consider combining them into a single operator. If it absolutely can’t be avoided, Airflow does have a feature for operator cross-communication called XCom that is described in the section XComs.

I fundamentally don't understand what they mean. If there's no data to pass between two tasks, why are they part of the same DAG?

I've got half a dozen different tasks that take turns editing one file in place, and each send an XML report to a final task that compiles a report of what was done. Airflow wants me to put all of that in one Operator? Then what am I gaining by doing it in Airflow? Or how can I restructure it in an Airflowy way?

Upvotes: 4

Views: 4476

Answers (3)

rescdsk
rescdsk

Reputation: 8895

I've decided that, as mentioned by @Anand Vidvat, they are making a distinction between Operators and Tasks here. What I think is that they don't want you to write two Operators that inherently need to be paired together and pass data to each other. On the other hand, it's fine to have one task use data from another, you just have to provide filenames etc in the DAG definition.

For example, many of the builtin Operators have constructor parameters for files, like the S3FileTransformOperator. Confusing documentation, but oh well!

Upvotes: 0

Javier Lopez Tomas
Javier Lopez Tomas

Reputation: 2352

What you have to do for avoiding having everything in one operator is saving the data somewhere. I don't quite understand your flow, but if for instance, you want to extract data from an API and insert that in a database, you would need to have:

  1. PythonOperator(or BashOperator, whatever) that takes the data from the API and saves it to S3/local file/Google Drive/Azure Storage...
  2. SqlRelated operator that takes the data from the storage and insert it into the database

Anyway, if you know which files are you going to edit, you may also use jinja templates or reading info from a text file and make a loop or something in the DAG. I could help you more if you clarify a little bit your actual flow

Upvotes: 3

Anand Vidvat
Anand Vidvat

Reputation: 1059

fundamentally, each instance of an operator in a DAG is mapped to a different task.

This is a subtle but very important point: in general if two operators need to share information, like a filename or small amount of data, you should consider combining them into a single operator

the above sentence means that if you want any information that needs to be shared between two different tasks then it is best you could combine them into one task instead of using two different tasks, on the other hand, if you must use two different tasks and you need to pass some information from one task to another then you can do it using Airflow's XCOM, which is similar to a key-value store.

In a Data Engineering use case, file schema before processing is important. imagine two tasks as follows :

  1. Files_Exist_Check : the purpose of this task is to check whether particular files exist in a directory or not before continuing.
  2. Check_Files_Schema: the purpose of this task is to check whether the file schema matches the expected schema or not.

It would only make sense to start your processing if Files_Exist_Check task succeeds. i.e. you have some files to process. In this case, you can "push" some key to xcom like "file_exists" with the value being the count of files present in that particular directory in Task Files_Exist_Check. Now, you "pull" this value using the same key in Check_Files_Schema Task, if it returns 0 then there are no files for you to process hence you can raise exception and fail the task or handle gracefully.

hence sharing information across tasks using xcom does come in handy in this case.

you can refer following link for more info :

  1. https://www.astronomer.io/guides/airflow-datastores/
  2. Airflow - How to pass xcom variable into Python function

Upvotes: 2

Related Questions