You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/03/31 11:10:43 UTC
[GitHub] [airflow] samhita-alla opened a new pull request #22646: Airflow flyte operator
samhita-alla opened a new pull request #22646:
URL: https://github.com/apache/airflow/pull/22646
Signed-off-by: Samhita Alla <aa...@gmail.com>
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
This PR adds a new community provider to allow Airflow users to interact with Flyte from within Airflow.
A bit about Flyte: [Flyte](https://www.flyte.org/) 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](https://github.com/flyteorg/flyte/issues/544).
Code Interface
---------------
The provider defines a `hook`, an `operator`, and a `sensor`. When the user instantiates the `AirflowFlyteOperator`, it creates a `FlyteRemote` (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 the `hook`. The executions are triggered synchronously by default. One can set `asynchronous` to `True` if the execution needs to be polled asynchronously; the polling would be handled by the `sensor`.
Tests
------
Unit tests have been added and tested. Also, I spun up an Airflow instance to validate the code in real-time.
TODO
------
- [ ] Update the flytekit version from `0.32.0b0` to `0.32.0` after the latter release is out.
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/main/UPDATING.md).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] raphaelauv commented on a change in pull request #22646: Add new community provider: Flyte
Posted by GitBox <gi...@apache.org>.
raphaelauv commented on a change in pull request #22646:
URL: https://github.com/apache/airflow/pull/22646#discussion_r839742314
##########
File path: airflow/providers/flyte/example_dags/example_flyte.py
##########
@@ -0,0 +1,77 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Example DAG demonstrating the usage of the AirflowFlyteOperator."""
+
+from datetime import datetime, timedelta
+
+from airflow import DAG
+from airflow.providers.flyte.operators.flyte import AirflowFlyteOperator
+from airflow.providers.flyte.sensors.flyte import AirflowFlyteSensor
+
+with DAG(
+ dag_id="example_flyte_operator",
+ schedule_interval=None,
+ start_date=datetime(2021, 1, 1),
+ dagrun_timeout=timedelta(minutes=60),
+ tags=["example"],
+ catchup=False,
+) as dag:
+
+ # [START howto_operator_flyte_synchronous]
+ sync_predictions = AirflowFlyteOperator(
Review comment:
why prefix all 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**
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] samhita-alla commented on a change in pull request #22646: Add new community provider: Flyte
Posted by GitBox <gi...@apache.org>.
samhita-alla commented on a change in pull request #22646:
URL: https://github.com/apache/airflow/pull/22646#discussion_r839485085
##########
File path: airflow/providers/flyte/operators/flyte.py
##########
@@ -0,0 +1,148 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import re
+from datetime import timedelta
+from typing import TYPE_CHECKING, Any, Dict, Optional, Sequence
+
+from airflow.exceptions import AirflowException
+from airflow.models import BaseOperator
+from airflow.providers.flyte.hooks.flyte import AirflowFlyteHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class AirflowFlyteOperator(BaseOperator):
+ """
+ Launch Flyte executions from within Airflow.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:AirflowFlyteOperator`
+
+ :param flyte_conn_id: Required. The connection to Flyte setup, containing metadata.
+ :param project: Optional. The project to connect to.
+ :param domain: Optional. The domain to connect to.
+ :param launchplan_name: Optional. The name of the launchplan to trigger.
+ :param task_name: Optional. The name of the task to trigger.
+ :param max_parallelism: Optional. The maximum number of parallel executions to allow.
+ :param raw_data_prefix: Optional. The prefix to use for raw data.
+ :param assumable_iam_role: Optional. The IAM role to assume.
+ :param kubernetes_service_account: Optional. The Kubernetes service account to use.
+ :param labels: Optional. Custom labels to be applied to the execution resource.
+ :param annotations: Optional. Custom annotations to be applied to the execution resource.
+ :param version: Optional. The version of the launchplan/task to trigger.
+ :param inputs: Optional. The inputs to the launchplan/task.
+ :param timeout: Optional. The timeout to wait for the execution to finish.
+ :param poll_interval: Optional. The interval between checks to poll the execution.
+ :param asynchronous: Optional. Whether to wait for the execution to finish or not.
+ """
+
+ template_fields: Sequence[str] = ("flyte_conn_id",) # mypy fix
+
+ def __init__(
+ self,
+ flyte_conn_id: str,
+ project: Optional[str] = None,
+ domain: Optional[str] = None,
+ launchplan_name: Optional[str] = None,
+ task_name: Optional[str] = None,
+ max_parallelism: Optional[int] = None,
+ raw_data_prefix: Optional[str] = None,
+ assumable_iam_role: Optional[str] = None,
+ kubernetes_service_account: Optional[str] = None,
+ labels: Dict[str, str] = {},
+ annotations: Dict[str, str] = {},
+ version: Optional[str] = None,
+ inputs: Dict[str, Any] = {},
+ timeout: Optional[timedelta] = None,
+ poll_interval: timedelta = timedelta(seconds=30),
+ asynchronous: bool = False,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.flyte_conn_id = flyte_conn_id
+ self.project = project
+ self.domain = domain
+ self.launchplan_name = launchplan_name
+ self.task_name = task_name
+ self.max_parallelism = max_parallelism
+ self.raw_data_prefix = raw_data_prefix
+ self.assumable_iam_role = assumable_iam_role
+ self.kubernetes_service_account = kubernetes_service_account
+ self.labels = labels
+ self.annotations = annotations
+ self.version = version
+ self.inputs = inputs
+ self.timeout = timeout
+ self.poll_interval = poll_interval
+ self.asynchronous = asynchronous
+ self.execution_name: str = ""
+
+ if (not (self.task_name or self.launchplan_name)) or (self.task_name and self.launchplan_name):
+ raise AirflowException("Either task_name or launchplan_name is required.")
+
+ def execute(self, context: "Context") -> str:
+ """Trigger an execution and wait for it to finish."""
+
+ # 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))]
Review comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] raphaelauv commented on a change in pull request #22646: Add new community provider: Flyte
Posted by GitBox <gi...@apache.org>.
raphaelauv commented on a change in pull request #22646:
URL: https://github.com/apache/airflow/pull/22646#discussion_r839742314
##########
File path: airflow/providers/flyte/example_dags/example_flyte.py
##########
@@ -0,0 +1,77 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Example DAG demonstrating the usage of the AirflowFlyteOperator."""
+
+from datetime import datetime, timedelta
+
+from airflow import DAG
+from airflow.providers.flyte.operators.flyte import AirflowFlyteOperator
+from airflow.providers.flyte.sensors.flyte import AirflowFlyteSensor
+
+with DAG(
+ dag_id="example_flyte_operator",
+ schedule_interval=None,
+ start_date=datetime(2021, 1, 1),
+ dagrun_timeout=timedelta(minutes=60),
+ tags=["example"],
+ catchup=False,
+) as dag:
+
+ # [START howto_operator_flyte_synchronous]
+ sync_predictions = AirflowFlyteOperator(
Review comment:
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**
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] samhita-alla commented on a change in pull request #22646: Add new community provider: Flyte
Posted by GitBox <gi...@apache.org>.
samhita-alla commented on a change in pull request #22646:
URL: https://github.com/apache/airflow/pull/22646#discussion_r840313009
##########
File path: airflow/providers/flyte/example_dags/example_flyte.py
##########
@@ -0,0 +1,77 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Example DAG demonstrating the usage of the AirflowFlyteOperator."""
+
+from datetime import datetime, timedelta
+
+from airflow import DAG
+from airflow.providers.flyte.operators.flyte import AirflowFlyteOperator
+from airflow.providers.flyte.sensors.flyte import AirflowFlyteSensor
+
+with DAG(
+ dag_id="example_flyte_operator",
+ schedule_interval=None,
+ start_date=datetime(2021, 1, 1),
+ dagrun_timeout=timedelta(minutes=60),
+ tags=["example"],
+ catchup=False,
+) as dag:
+
+ # [START howto_operator_flyte_synchronous]
+ sync_predictions = AirflowFlyteOperator(
Review comment:
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!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] samhita-alla commented on a change in pull request #22646: Airflow flyte operator
Posted by GitBox <gi...@apache.org>.
samhita-alla commented on a change in pull request #22646:
URL: https://github.com/apache/airflow/pull/22646#discussion_r839485085
##########
File path: airflow/providers/flyte/operators/flyte.py
##########
@@ -0,0 +1,148 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import re
+from datetime import timedelta
+from typing import TYPE_CHECKING, Any, Dict, Optional, Sequence
+
+from airflow.exceptions import AirflowException
+from airflow.models import BaseOperator
+from airflow.providers.flyte.hooks.flyte import AirflowFlyteHook
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class AirflowFlyteOperator(BaseOperator):
+ """
+ Launch Flyte executions from within Airflow.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:AirflowFlyteOperator`
+
+ :param flyte_conn_id: Required. The connection to Flyte setup, containing metadata.
+ :param project: Optional. The project to connect to.
+ :param domain: Optional. The domain to connect to.
+ :param launchplan_name: Optional. The name of the launchplan to trigger.
+ :param task_name: Optional. The name of the task to trigger.
+ :param max_parallelism: Optional. The maximum number of parallel executions to allow.
+ :param raw_data_prefix: Optional. The prefix to use for raw data.
+ :param assumable_iam_role: Optional. The IAM role to assume.
+ :param kubernetes_service_account: Optional. The Kubernetes service account to use.
+ :param labels: Optional. Custom labels to be applied to the execution resource.
+ :param annotations: Optional. Custom annotations to be applied to the execution resource.
+ :param version: Optional. The version of the launchplan/task to trigger.
+ :param inputs: Optional. The inputs to the launchplan/task.
+ :param timeout: Optional. The timeout to wait for the execution to finish.
+ :param poll_interval: Optional. The interval between checks to poll the execution.
+ :param asynchronous: Optional. Whether to wait for the execution to finish or not.
+ """
+
+ template_fields: Sequence[str] = ("flyte_conn_id",) # mypy fix
+
+ def __init__(
+ self,
+ flyte_conn_id: str,
+ project: Optional[str] = None,
+ domain: Optional[str] = None,
+ launchplan_name: Optional[str] = None,
+ task_name: Optional[str] = None,
+ max_parallelism: Optional[int] = None,
+ raw_data_prefix: Optional[str] = None,
+ assumable_iam_role: Optional[str] = None,
+ kubernetes_service_account: Optional[str] = None,
+ labels: Dict[str, str] = {},
+ annotations: Dict[str, str] = {},
+ version: Optional[str] = None,
+ inputs: Dict[str, Any] = {},
+ timeout: Optional[timedelta] = None,
+ poll_interval: timedelta = timedelta(seconds=30),
+ asynchronous: bool = False,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.flyte_conn_id = flyte_conn_id
+ self.project = project
+ self.domain = domain
+ self.launchplan_name = launchplan_name
+ self.task_name = task_name
+ self.max_parallelism = max_parallelism
+ self.raw_data_prefix = raw_data_prefix
+ self.assumable_iam_role = assumable_iam_role
+ self.kubernetes_service_account = kubernetes_service_account
+ self.labels = labels
+ self.annotations = annotations
+ self.version = version
+ self.inputs = inputs
+ self.timeout = timeout
+ self.poll_interval = poll_interval
+ self.asynchronous = asynchronous
+ self.execution_name: str = ""
+
+ if (not (self.task_name or self.launchplan_name)) or (self.task_name and self.launchplan_name):
+ raise AirflowException("Either task_name or launchplan_name is required.")
+
+ def execute(self, context: "Context") -> str:
+ """Trigger an execution and wait for it to finish."""
+
+ # 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))]
Review comment:
'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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] samhita-alla commented on pull request #22646: Add new community provider: Flyte
Posted by GitBox <gi...@apache.org>.
samhita-alla commented on pull request #22646:
URL: https://github.com/apache/airflow/pull/22646#issuecomment-1085530452
Hi, @potiuk! Started a discussion: https://lists.apache.org/thread/b55g3gydgmqmhow6f7xzzbm5t0gmhs2x.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #22646: Add new community provider: Flyte
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #22646:
URL: https://github.com/apache/airflow/pull/22646#issuecomment-1084509470
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #22646: Add new community provider: Flyte
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #22646:
URL: https://github.com/apache/airflow/pull/22646#issuecomment-1084509906
I am not telling we do not want Flyte, but I think it should be discussed there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org