-
Notifications
You must be signed in to change notification settings - Fork 14.3k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add new community provider: Flyte #22646
Conversation
Signed-off-by: Samhita Alla <[email protected]>
Signed-off-by: Samhita Alla <[email protected]>
Signed-off-by: Samhita Alla <[email protected]>
Signed-off-by: Samhita Alla <[email protected]>
# create a deterministic execution name | ||
task_id = re.sub(r"[\W_]+", "", context["task"].task_id)[:5] | ||
self.execution_name = task_id + re.sub( | ||
r"[\W_]+", | ||
"", | ||
context["dag_run"].run_id.split("__")[-1].lower(), | ||
)[: (20 - len(task_id))] |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'm currently generating a deterministic-and-unique execution name that is to be used to name a Flyte execution. It's a combination of the task_id and run_id. I'm using both because with task_id, I wouldn't be able to create unique task names whenever a task runs more than once, and with run_id I wouldn't be able to create unique task names within the same DAG cause run_id remains the same for all tasks; hence came up with this logic.
The execution name cannot exceed 20 characters (a restriction imposed by Flyte), and hence, I'm trimming the two strings. Please let me know if there's a better way to create unique execution names, even with the task being repeated multiple times within the same DAG (with different task_ids, of course), or run multiple times.
Signed-off-by: Samhita Alla <[email protected]>
Hello @samhita-alla - I think you might want to read, what I wrote in https://lists.apache.org/thread/t1k3d0518v4kxz1pqsprdc78h0wxobg0 and possibly chime-in or start a new discussion in devliust about contributing Flyte. I think adding new provider should now always be discussed on the devlist first. |
I am not telling we do not want Flyte, but I think it should be discussed there. |
) as dag: | ||
|
||
# [START howto_operator_flyte_synchronous] | ||
sync_predictions = AirflowFlyteOperator( |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
why prefix your operators with Airflow
?
example KubernetesPodOperator
is not AirflowKubernetesPodOperator
also FlyteOperator
is not explicit , I guess it's to trigger a run in Flyte
then a good name could be FlyteRunOperator
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We've prefixed the name with "Airflow" because we want to explicitly tell what the operator is. But, as you mentioned, it might not be necessary. We'll think about it. Thanks for looking into the PR!
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
+1. we should drop the Airflow as the prefix
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yep. +1
Hi, @potiuk! Started a discussion: https://lists.apache.org/thread/b55g3gydgmqmhow6f7xzzbm5t0gmhs2x. |
Hello @samhita-alla - as a follow up (as we have 5(!) new provider PRs submitted for review) I started a more "general" discussion on our approach for providers: https://lists.apache.org/thread/nvfc75kj2w1tywvvkw8ho5wkx1dcvgrn As such - I would really love if you comment there - for now there are different voices and I think I would love to hear from the companies that submit new providers, what is really their goal and why they think submitting the provider to Airflow community is better than maintaining the provider on their own. So far in the discussions it seems that there is no real incentive for the comunity to bring more providers in and our long term plans fit better with services like Flyte mainitaining and releasing their providers on their own (also without having any constraints that becoming part of the community ASF-managed code brings). But maybe we fail to see the reasoning from your side, so we would very much like to see what are your goals and reasoning behind submitting code to Airflow and what you want to achieve with that - and maybe we can work together with various servides to fulfill your goals without you having to submit code to Airflow. Please - comment there - we need your input there. |
Please find my comment here: https://lists.apache.org/thread/7ddgm6fbrzgf0bjww4hrh1bmgjz0c052 |
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions. |
not stale yet. We will come back to the discussion soon. |
Hello @samhita-alla - It took a bit long but I think we got to a consensus regarding new providers, and we also propose some kind of mix-governance aproach, where (among the others) the stakeholders for the future providers (which we are going to technically split to separate repositories soon) should take a bit more responsibility for maintenance: If that does not scare you away, and you still want to add the provider to Airflow community providers, feel free to rebase the PR. I will also ask you (hopefully it will be merged soon) to rebase it after we merge #24672 - we are going to change the way how we keep depdencies for providers in order to prepare them to separate to different repository. Let us know what you want to do, either close the PR or rebase it and lead it to completion after #24672 is merged. |
Hi, @potiuk! Thanks for letting me know about the updated process. I have, however, created a standalone repository already to house the Flyte Airflow Provider which I'm planning to publish to the Astronomer registry. So lemme close this PR. Thanks for keeping me in the loop, though! |
Cool. I think it's a good choice :) |
Good to see more community providers, thank you @samhita-alla - let us know if you need any help publishing it with the Astronomer registry |
@kaxil, sure thing, thanks! |
Signed-off-by: Samhita Alla [email protected]
This PR adds a new community provider to allow Airflow users to interact with Flyte from within Airflow.
A bit about Flyte: Flyte is an open-source, container-native, structured programming and distributed processing platform that enables highly concurrent, scalable, and maintainable workflows for machine learning and data processing pipelines.
As a more significant chunk of the users who are into pipelines are using Airflow, it'd be really helpful to have a provider that bridges the gap between Airflow and Flyte, to help the Airflow users retain their existing pipelines and use Flyte from within the Airflow DAGs to run the machine learning jobs (say).
We've had this operator in the back of our minds for a long time; here's the issue.
Code Interface
The provider defines a
hook
, anoperator
, and asensor
. When the user instantiates theAirflowFlyteOperator
, it creates aFlyteRemote
(Flyte's Python API) object, triggers the execution in the Flyte environment, waits for the execution to complete, and finally, returns the execution name. This implementation is carried out using the methods defined in thehook
. The executions are triggered synchronously by default. One can setasynchronous
toTrue
if the execution needs to be polled asynchronously; the polling would be handled by thesensor
.Tests
Unit tests have been added and tested. Also, I spun up an Airflow instance to validate the code in real-time.
TODO
0.32.0b0
to0.32.0
after the latter release is out.^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.