API Endpoints for Dataset Creation and Updating

[edumuellerFSL]

Description

I would like to propose the addition of new API endpoints for creating and updating datasets in Airflow. This feature would be a valuable extension to the current dataset capabilities and would align with the direction Airflow is heading, especially considering the dataset listeners introduced in Airflow 2.8.

Proposed Changes:

1. Addition of Dataset Create API Endpoint: This endpoint will enable users to create new datasets directly via API.
2. Addition of Dataset Update API Endpoint: This endpoint will allow users to update existing datasets via API to trigger a change.

Use case/motivation

In a multi-instance Airflow architecture, managing dataset dependencies across instances can be challenging, as we are currently experiencing in our organization.

This feature also aligns with the recent advancements in Airflow 2.8, particularly with the introduction of dataset listeners. These developments have opened the door for improved cross-instance dataset awareness, an area where this proposal would be extremely beneficial.

We believe that with the introduction of these new endpoints, Airflow would offer a more efficient and facilitated approach to cross-instance dataset-aware scheduling. This enhancement would not only benefit our organization but also the broader Airflow community, as it is likely a common challenge faced by many and more will likely encounter in the future.

Related issues

This feature complements the discussions and contributions already seen in the community, especially those related to enhancing dataset management and integration in Airflow.

There have been some ongoing discussions and contributions on GitHub, e.g. #36308 #29162, including a previously closed Pull Request (#29433).

These discussions highlight the community's interest in and need for enhanced dataset management capabilities.

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[naltimari]

I also believe this would be helpful and aligns with the data-aware scheduling feature, as well as the dataset listener feature.

[potiuk]

I think it is far too big of a feature and should be discussed at the devlist. Currently all the objects (DAGs and Datasaets alike) are created by parsing DAG files, NOT by creating DB entities. IMHO it makes very little sense to start creating those datasets via APIs. Especially that Datasets are not "standalone" entities and that there is nothing that could happen if you create a dataset via API but there is no DAG file that uses it.

I am not sure what would be the consequences of it as I know there are other discussions happening about dataset future - but I think if you want to start anything about that, starting a devlist discussion and explaining what you want is really the right way of approaching it.

Converting it into discussion as this is definitely not a "feature" scope.

[edumuellerFSL]

I have moved the reply to a comment since it's a bit lengthy.

[potiuk]

See https://airflow.apache.org/community/ for devlist information.

[edumuellerFSL]

Hi @potiuk and Airflow community,

Thank you, @potiuk for you insights. I understand the concerns about deviating from the current method of creating objects via DAG file parsing. However, I'd like to briefly outline our use case in a multi-tenant Airflow setup and seek your guidance or alternative solutions.

In our organization, we're dealing with multiple Airflow tenants, let's say Tenant 1 and Tenant 2, as examples. Our goal is to create a more interconnected workflow between these tenants, specifically focusing on Datasets.

Our Current Setup

• We operate in a multi-tenant Airflow environment (e.g., Tenant 1 and Tenant 2).
• To synchronize Dataset A across these tenants, we currently have a complex setup:
  • Containers run on a schedule to export metadata to CosmosDB (these will be replaced by the listener).
  • Additional scheduled containers pull data from CosmosDB and write it to a shared file system, enabling generated DAGS to read it and mirror a dataset across tenants.*

Proposed Workflow

Here's a breakdown of our proposed workflow:

1. Cross-Tenant Dataset Interaction: We have Dags in Tenant 1 producing Dataset A. We need a mechanism to trigger all Dags consuming Dataset A in Tenant 2. This interaction is crucial for our data pipeline's efficiency and synchronicity.
2. Dataset Listener Implementation: Our approach involves implementing a Dataset listener that programmatically creates Dataset A in all tenants where it's not present (like Tenant 2) and export Dataset updates when they happen. This would trigger an update on all Dags consuming from that Dataset.
3. Standardized Dataset Names: We plan to use standardized dataset names, which makes sense since a URI is its identifier and uniqueness is a logical requirement.

Why This Matters:

• It offers a streamlined, automated way to manage datasets across different Airflow instances.
• It aligns with our need for efficient, interconnected workflows in a multi-tenant environment.

Do you see an alternative method within the current Airflow framework that could achieve this goal? We're open to different approaches as long as they provide a practical solution for our use case.

I will be sending a proposal to the devlist so this can be discussed further, as suggested.

Thank you again for your time considering this proposal, and I look forward to any suggestions or guidance you might have.

[dantheman0207]

A common use case at our company is synchronizing datasets with other technologies (as @cmarteepants mentioned). Creating a Dataset via API would allow us to keep Airflow in sync with other tools and teams whose technical choices we can't control. This would let us use Airflow more and simplify internal collaboration.

[RNHTTR]

how does creating a random dataset that no task produces or dag schedules against help at all?

@jedcunningham I suppose it doesn't :) I mistakenly thought you needed both a producer & consumer for a dataset to be created.

[Blizzke]

A common use case at our company is synchronizing datasets with other technologies (as @cmarteepants mentioned). Creating a Dataset via API would allow us to keep Airflow in sync with other tools and teams whose technical choices we can't control. This would let us use Airflow more and simplify internal collaboration.

@dantheman0207 I'm curious to see how you see the "sync with other tools" working if it's not triggered by a DAG/Task? Are you planning to read from the airflow database directly?

We also need this functionality, but I personally don't see added value in keeping datasets in airflow that are not connected to anything. I mean, if a dataset is not triggering a DAG/Task, or it is not produced as the result of a DAG/Task, what's its use? Airflow is not a data lineage tool (although I do enjoy the graph :-) )

[cmarteepants]

@RNHTTR I actually don't (yet! 😆) have an opinion on whether or not we should allow creating datasets via API, but I do understand why it's controversial and I think going to the dev list to talk about whether or not we want to go down that road is is the right call.

Updating Datasets (i.e creating dataset events via API) however seems like a no brainer and in of itself unlocks some new capabilities. My only goal was to separate this into two different asks, and encourage someone to pick up the latter.

Also:

Airflow is not a data lineage tool (although I do enjoy the graph :-) )

But it could be 😈

[potiuk]

But it could be 😈

Airflow already exposes Lineage information through integrated Open Lineage API. And this is as far it will go. This is the embodiment of "Airflow As a Platform" mantra. And this is the strategic direction of the community as far as I can tell. It PRODUCES the lineage data but our goal is not to OWN it.

[funes79]

I would like just to mention that also in one-airflow environment this (i.e. possibility to update, add, Dataset event via API, web... whatever) can be very helpful. In cases where you run "Daily" dataset scheduled DAGs - see my issue and question here: #36618

Or - in my case the possibility to edit DAG dataset schedule time (i.e. last time what is considered for the datasets to be presented as ready) would be also helpful.

[Blizzke]

We were eagerly waiting for the MR as it would solve a lot of our problems.

We have a very large flow of data in our company, of which a big part is still being handled "the old fashioned way".
Right now a lot of our DAGs have sensors or have to be triggered via the API externally every time an external workflow is ready.

Also, we often have more than one DAG awaiting the end of an external workflow and that means we have to put a DAG in between to trigger all the related DAGs. Because we don't control the code that does the API triggering, we'd have to ask for a cross-team code update every time a new DAG was added (if we didn't add the in-between DAG).

All this to say: While this works, it is a mess to maintain.

Having external Dataset support would be a blessing.
The external workflows could just register a dataset update and everything would work.
Internally, all we'd have to do is modify the scheduler to watch the dataset and presto.
Plus the added benefits of the nice graphs and lineage :-)

I get that there is a big discussion about this, it is a huge change, but for our (the companies) way of working, being able to use datasets has made life so much easier.

As a remark on the side (not related to this discussion, but I wanted to share): We've also overridden DatasetManager:register_dataset_change() . If the extra argument is None, but the referenced Datasets' extra is not empty, it should use the Datasets' extra for the change in our opinion.

[jedcunningham]

So in your situation, the piece you really want is to be able to register a dataset update via api, correct? Do you care about creating new datasets via api?

[Blizzke]

No, we are looking for the same behavior as internally. If a register-dataset-change-event comes in via the API that has no DAGs listening for it, we couldn't care less that it is ignored. We just want to extend the use of datasets to our external tooling.

The call should not result in an error though. I mean, "not relevant" is not really an error anyway right? So it should just be ignored then. (The external flows usually use the exit codes of the calls they do to determine to continue or not.

[Blizzke]

For people wanting to use something now already, we've been emulating this functionality via a DAG that receives a set of Datasets as a string parameter (JSON) and adds them to the task as outlets.
We trigger this via de regular dagRuns API to start it.
It has pretty much the same end result, albeit a bit less clean 🙂

It looks pretty much like this:

with DAG(
        "datasets-dispatcher",
       ....
        params={
            'datasets': Param('', description='Datasets (JSON list)', type='string')
        }
):
    @task(task_id=f"dispatch")
    def dispatcher(params: dict = None, task=None):
        from json import loads

        datasets = (params or {}).get('datasets', [])
        if datasets:
            datasets = loads(datasets)

        task.add_outlets([Dataset(**ds) for ds in datasets])

    dispatcher()

The received JSON for the datasets parameter is basically just a [{"uri": "...", "extra": {...}}, ...]

[potiuk]

See the message below. This is a serious change in the direction of what datasets are supposed to do and the only way it can be implemented is the discussion, followed by either a CONSENSUIS or VOTE on the devlist. Mere presence of a PR does not mean much. There were plenty of ideas and PRs that did not pass the community approval and were rejected in the past (and some of them were my PRs), so it's not ME personally who has to make those decisions, you need to get community approval if you want it in as I explained in the other message).

[Blizzke]

Yes but since I'm not an Airflow DEV I cannot participate in that discussion.

May I also point out that, if I understand correctly and the plan is to remove the dynamic definition of datasets in the future (and demand that they are defined at DAG level itself), for us this will break a lot of processes.
All our ETLs run in containers on other platforms and the datasets they consume and generate are extracted out of those containers and added to the taskinstance in the same way as the DAG above (using the TI inlets/outlets functionality). It is technically not possible for us to pre-define all Datasets at DAG level in advance. So please consider those things as well before you do.

[potiuk]

Yes but since I'm not an Airflow DEV I cannot participate in that discussion.

It's a matter of subscribing to the devlist. There is no separate airflow DEV. Airflow GitHub has now more than 2800 and every single one is Airflow DEV even if they contributed one line change.

Devlist is where important discussion happen and devlist is where decisions are made. Full stop.

All the discussions here are just people talking about their ideas. There is no other significance, no other weight to those dicsussions. Various people in the community might see and comment on it, but many of those who will be voting won't even know about this discussion happening unless it is brought to the devlist. And eventually those people COMMITTERS are going to vote on it.

So if you have an idea that is controversial, my suggestion is to participate in the devlist discussion and convince those COMMITTERS that what you propose makes sense.

Airflow is an Apache Software Foundation project and there are very clear rules about Governance, decisions making and approvals. You might not like it, but this is the reality you are facing, so I suggest you follow that if you want something to happen - otherwise it's an empty threadmill running.

[cmarteepants]

The Airflow Community Page has instructions on joining the dev list, and anybody can join. @Blizzke Send an email to [email protected] and come join the fun 😄

[nitin1677]

For people wanting to use something now already, we've been emulating this functionality via a DAG that receives a set of Datasets as a string parameter (JSON) and adds them to the task as outlets. We trigger this via de regular dagRuns API to start it. It has pretty much the same end result, albeit a bit less clean 🙂

It looks pretty much like this:

with DAG(
        "datasets-dispatcher",
       ....
        params={
            'datasets': Param('', description='Datasets (JSON list)', type='string')
        }
):
    @task(task_id=f"dispatch")
    def dispatcher(params: dict = None, task=None):
        from json import loads

        datasets = (params or {}).get('datasets', [])
        if datasets:
            datasets = loads(datasets)

        task.add_outlets([Dataset(**ds) for ds in datasets])

    dispatcher()

The received JSON for the datasets parameter is basically just a [{"uri": "...", "extra": {...}}, ...]

can you please share some more details for this implementation? is there any working example on github or other channel.

[edumuellerFSL]

Hello everyone. I appreciate the diverse perspectives shared in this discussion on API endpoints for dataset creation and updating in Airflow. Let me address some of the key points raised:

@potiuk, I recognize the significance of the traditional method of defining DAG objects via code and agree that any deviation from this needs careful consideration. However, as @cmarteepants and others have highlighted, there's a growing need for Airflow to adapt to more dynamic and interconnected workflows, particularly in multi-tenant environments.

Agreeing with many here, an API endpoint to trigger Dataset Updates fits well with Airflow's existing features and addresses a critical need for more flexible scheduling. As @Blizzke and @funes79 pointed out, the ability to handle dataset events via API would greatly simplify and improve the efficiency of workflows that depend on external data sources or cross-team interactions, not to mention the impact it will have on facilitating synchronization with other technologies. Focusing on the update endpoint first seems sensible given its broader acceptance.

On creating datasets via API, despite some reservations raised by @jedcunningham and @RNHTTR, I think it's a path worth exploring. As @dantheman0207 points out, the ability to sync datasets across different technologies is essential, and API creation of datasets could significantly boost Airflow's integration power and workflow orchestration.

Here's my take on this: Data-aware scheduling was a transformative step for Airflow because it acknowledged data as the primary workflow trigger. This proposal is essentially an extension of that concept, aiming to further decouple Airflow from the assumption that only DAGs can influence datasets.

Strategically, it would enable Airflow to serve a wider array of use cases and technological setups, allowing it to embrace the diversity of data sources and workflows, something particularly interesting and necessary for larger organizations.

[potiuk]

I strongly recommend to continue (and keep) that discussion in devlist https://lists.apache.org/thread/q4sd09xymvlskxdgcb1s6gfsqwnvqs2o

There is no way me or any other single person can take decision here (and they will not be taken in here nor in PRs).

My strong advice here is stop discussing it here or in PRs, but focus on getting feedback (and encouraging it) from the communityu in the devlist discussion.

This is a strategic decision on where Airflow Dataset should go. Different people have - apparently different opinions here. So (as is the case in Airflow being ASF project) the decision has to be made on the devlist, and in case consensus cannot be reached, if somoene would like to go in that direction, they will have to put it up for a VOTE (after sufficient time is given for people to discuss it and a number of community members have a say there).

See https://www.apache.org/foundation/voting.html for the rules about the voting that you will likely have to run to get it approved by the community. This is a "Code modification" rule, so I think before you put it up to a vote, if you want it in, you need to be reasonably sure that there is no Veto from a committer.

I have not yet made up my mind, I would love to hear what those who created datasets and who (I know they do) think about the future of datasets - have to say, but my feeling so far is that this is against the way we designed it and abusing Airflow for something other tools should do. But I would like to hear more from others.

If I may suggest also - if you want to present it and your arguments in a more "live" fashion - we have a monthly Airflow Town Hall Meetup.. This is a new thing we are doing, but I think it might be a good idea for you to use the opportunity to present your idea - if you want to advocate for it. Simply send (again - in devlist) for a presentation slot - look for "Town Hall" in the devlist archives and respond to it. Note that this meeting is merely to share things, no decisions are made there. The only place such decisions can be made are in devlist.

[nitin1677]

we also need something like this feature with multi tenant environments and publish dataset events for targeted user group for a given environment via rest api. as targeted user groups are dynamic for each environment and can be created/added periodically in system of record, if dataset not exists, creating dataset event with post api fails with error dataset not exists.