You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by js...@apache.org on 2025/01/05 21:58:47 UTC
(airflow) branch main updated: Document deployment of Edge Worker on Windows (#45403)
This is an automated email from the ASF dual-hosted git repository.
jscheffl pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 03659e4ece3 Document deployment of Edge Worker on Windows (#45403)
03659e4ece3 is described below
commit 03659e4ece38ef82f26b4c797ec053f0462c6324
Author: Jens Scheffler <95...@users.noreply.github.com>
AuthorDate: Sun Jan 5 22:58:39 2025 +0100
Document deployment of Edge Worker on Windows (#45403)
* Document deployment of Edge Worker on Windows
* Spelling
* Review
---
.../src/airflow_breeze/params/shell_params.py | 11 +
.../edge_executor.rst | 16 +-
docs/apache-airflow-providers-edge/index.rst | 1 +
.../install_on_windows.rst | 94 ++++++
.../edge/example_dags/integration_test.py | 4 +-
.../providers/edge/example_dags/win_notepad.py | 83 +++++
.../providers/edge/example_dags/win_test.py | 339 +++++++++++++++++++++
scripts/in_container/bin/run_tmux | 3 +-
8 files changed, 540 insertions(+), 11 deletions(-)
diff --git a/dev/breeze/src/airflow_breeze/params/shell_params.py b/dev/breeze/src/airflow_breeze/params/shell_params.py
index 68bd75692ee..29049cf85d5 100644
--- a/dev/breeze/src/airflow_breeze/params/shell_params.py
+++ b/dev/breeze/src/airflow_breeze/params/shell_params.py
@@ -512,6 +512,17 @@ class ShellParams:
_env, "AIRFLOW__CORE__EXECUTOR", "airflow.providers.edge.executors.edge_executor.EdgeExecutor"
)
_set_var(_env, "AIRFLOW__EDGE__API_ENABLED", "true")
+
+ # For testing Edge Worker on Windows... Default Run ID is having a colon (":") from the time which is
+ # made into the log path template, which then fails to be used in Windows. So we replace it with a dash
+ _set_var(
+ _env,
+ "AIRFLOW__LOGGING__LOG_FILENAME_TEMPLATE",
+ "dag_id={{ ti.dag_id }}/run_id={{ ti.run_id|replace(':', '-') }}/task_id={{ ti.task_id }}/"
+ "{% if ti.map_index >= 0 %}map_index={{ ti.map_index }}/{% endif %}"
+ "attempt={{ try_number|default(ti.try_number) }}.log",
+ )
+
# Dev Airflow 3 runs API on FastAPI transitional
port = 9091
if self.use_airflow_version and self.use_airflow_version.startswith("2."):
diff --git a/docs/apache-airflow-providers-edge/edge_executor.rst b/docs/apache-airflow-providers-edge/edge_executor.rst
index 04286cc404c..f7c08ed6f81 100644
--- a/docs/apache-airflow-providers-edge/edge_executor.rst
+++ b/docs/apache-airflow-providers-edge/edge_executor.rst
@@ -20,19 +20,19 @@ Edge Executor
.. note::
- The Edge Provider Package is an experimental preview. Features and stability is limited
- and needs to be improved over time. Target is to have full support in Airflow 3.
- Once Airflow 3 support contains Edge Provider, maintenance of the Airflow 2 package will
- be dis-continued.
+ The Edge Provider Package is an experimental preview. Features and stability are limited,
+ and need to be improved over time. The target is to achieve full support in Airflow 3.
+ Once the Edge Provider is fully supported in Airflow 3, maintenance of the Airflow 2 package will
+ be discontinued.
.. note::
- As of Airflow 2.10.3, the ``edge`` provider package is not included in normal release cycle.
- Thus you can not directly install it via: ``pip install 'apache-airflow[edge]'`` as the dependency
- can not be downloaded.
+ As of Airflow 2.10.4, the ``edge`` provider package is not included in normal release cycle.
+ Thus, it cannot be directly installed using: ``pip install 'apache-airflow[edge]'`` as the dependency
+ cannot be downloaded.
- While it is in not-ready state, a wheel release package must be manually built from source tree
+ While it is in a not-ready state, a wheel release package must be manually built from source tree
via ``breeze release-management prepare-provider-packages --include-not-ready-providers edge``
and then installed via pip or uv from the generated wheel file. like:
``pip install apache_airflow_providers_edge-<version>-py3-none-any.whl``.
diff --git a/docs/apache-airflow-providers-edge/index.rst b/docs/apache-airflow-providers-edge/index.rst
index 36841578da9..86e011c08a2 100644
--- a/docs/apache-airflow-providers-edge/index.rst
+++ b/docs/apache-airflow-providers-edge/index.rst
@@ -28,6 +28,7 @@
Home <self>
Changelog <changelog>
Security <security>
+ Installation on Windows <install_on_windows>
.. toctree::
diff --git a/docs/apache-airflow-providers-edge/install_on_windows.rst b/docs/apache-airflow-providers-edge/install_on_windows.rst
new file mode 100644
index 00000000000..3e96e6eff9e
--- /dev/null
+++ b/docs/apache-airflow-providers-edge/install_on_windows.rst
@@ -0,0 +1,94 @@
+ .. 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.
+
+Install Edge Worker on Windows
+==============================
+
+.. note::
+
+ The Edge Provider Package is an experimental preview. Features and stability are limited,
+ and need to be improved over time. The target is to achieve full support in Airflow 3.
+ Once the Edge Provider is fully supported in Airflow 3, maintenance of the Airflow 2 package will
+ be discontinued.
+
+ This is also especially true for Windows. The Edge Worker has only been manually tested on Windows,
+ and the setup is not validated in CI. It is recommended to use Linux for Edge Worker. The
+ Windows-based setup is intended solely for testing at your own risk. It is technically limited
+ due to Python OS restrictions and if currently of Proof-of-Concept quality.
+
+
+.. note::
+
+ As of Airflow 2.10.4, the ``edge`` provider package is not included in normal release cycle.
+ Thus, it cannot be directly installed using: ``pip install 'apache-airflow[edge]'`` as the dependency
+ cannot be downloaded.
+
+ While it is in a not-ready state, a wheel release package must be manually built from source tree
+ via ``breeze release-management prepare-provider-packages --include-not-ready-providers edge``.
+
+
+The setup was tested on Windows 10 with Python 3.12.8, 64-bit.
+To setup a instance of Edge Worker on Windows, you need to follow the steps below:
+
+1. Install Python 3.9 or higher.
+2. Create an empty folder as base to start with. In our example it is ``C:\\Airflow``.
+3. Start Shell/Command Line in ``C:\\Airflow`` and create a new virtual environment via: ``python -m venv venv``
+4. Activate the virtual environment via: ``venv\\Scripts\\activate.bat``
+5. Copy the manually built wheel of the edge provider to the folder ``C:\\Airflow``.
+ This document used ``apache_airflow_providers_edge-0.9.7rc0-py3-none-any.whl``.
+6. Install the wheel file with the Airflow constraints matching your Airflow and Python version:
+ ``pip install apache_airflow_providers_edge-0.9.7rc0-py3-none-any.whl apache-airflow==2.10.4 virtualenv --constraint https://raw.githubusercontent.com/apache/airflow/constraints-2.10.4/constraints-3.12.txt``
+7. Create a new folder ``dags`` in ``C:\\Airflow`` and copy the relevant DAG files in it.
+ (At least the DAG files which should be executed on the edge alongside the dependencies. For testing purposes
+ the DAGs from the ``apache-airflow`` repository can be used located in
+ <https://github.com/apache/airflow/tree/main/providers/src/airflow/providers/edge/example_dags>.)
+8. Collect needed parameters from your running Airflow backend, at least the following:
+
+ - ``edge`` / ``api_url``: The HTTP(s) endpoint where the Edge Worker connects to
+ - ``core`` / ``internal_api_secret_key``: The shared secret key between the webserver and the Edge Worker
+ Note: This only applies to Airflow 2.10 - For Airflow 3 the authentication method might change before release.
+ - Any proxy details if applicable for your environment.
+
+9. Create a worker start script to prevent repeated typing. Create a new file ``start_worker.bat`` in
+ ``C:\\Airflow`` with the following content - replace with your settings:
+
+.. code-block:: bash
+
+ @echo off
+ set AIRFLOW__CORE__DAGS_FOLDER=dags
+ set AIRFLOW__LOGGING__BASE_LOG_FOLDER=edge_logs
+ set AIRFLOW__EDGE__API_URL=https://your-hostname-and-port/edge_worker/v1/rpcapi
+ set AIRFLOW__CORE__EXECUTOR=airflow.providers.edge.executors.edge_executor.EdgeExecutor
+ set AIRFLOW__CORE__INTERNAL_API_SECRET_KEY=<steal this from your deployment...>
+ set AIRFLOW__CORE__LOAD_EXAMPLES=False
+ set AIRFLOW_ENABLE_AIP_44=true
+ @REM Add if needed: set http_proxy=http://my-company-proxy.com:3128
+ @REM Add if needed: set https_proxy=http://my-company-proxy.com:3128
+ airflow edge worker --concurrency 4 -q windows
+
+10. Note on logs: Per default the DAG Run ID is used as path in the log structure and per default the date and time
+ is contained in the Run ID. Windows fails with a colon (":") in a file or folder name and this also
+ the Edge Worker fails.
+ Therefore you might consider changing the config ``AIRFLOW__LOGGING__LOG_FILENAME_TEMPLATE`` to avoid the colon.
+ For example you could add the Jinja2 template replacement ``| replace(":", "-")`` to use other characters.
+ Note that the log filename template is resolved on server side and not on the worker side. So you need to make
+ this as a global change.
+ Alternatively for testing purposes only you must use Run IDs without a colon, e.g. set the Run ID manually when
+ starting a DAG run.
+11. Start the worker via: ``start_worker.bat``
+ Watch the console for errors.
+12. Run a DAG as test and see if the result is as expected.
diff --git a/providers/src/airflow/providers/edge/example_dags/integration_test.py b/providers/src/airflow/providers/edge/example_dags/integration_test.py
index 128da9a8401..ba42665ddf7 100644
--- a/providers/src/airflow/providers/edge/example_dags/integration_test.py
+++ b/providers/src/airflow/providers/edge/example_dags/integration_test.py
@@ -47,8 +47,8 @@ with DAG(
description=__doc__.partition(".")[0],
doc_md=__doc__,
schedule=None,
- start_date=datetime(2024, 7, 1),
- tags=["example", "params", "integration test"],
+ start_date=datetime(2025, 1, 1),
+ tags=["example", "edge", "integration test"],
params={
"mapping_count": Param(
4,
diff --git a/providers/src/airflow/providers/edge/example_dags/win_notepad.py b/providers/src/airflow/providers/edge/example_dags/win_notepad.py
new file mode 100644
index 00000000000..da50fff8b96
--- /dev/null
+++ b/providers/src/airflow/providers/edge/example_dags/win_notepad.py
@@ -0,0 +1,83 @@
+#
+# 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.
+"""
+In this DAG is a demonstrator how to interact with a Windows Worker via Notepad.
+
+The DAG is created in conjunction with the documentation in
+https://github.com/apache/airflow/blob/main/docs/apache-airflow-providers-edge/install_on_windows.rst
+and serves as a PoC test for the Windows worker.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from datetime import datetime
+from pathlib import Path
+from subprocess import check_call
+from tempfile import gettempdir
+from typing import TYPE_CHECKING, Any
+
+from airflow.models import BaseOperator
+from airflow.models.dag import DAG
+from airflow.models.param import Param
+
+if TYPE_CHECKING:
+ from airflow.utils.context import Context
+
+
+class NotepadOperator(BaseOperator):
+ """Example Operator Implementation which starts a Notepod.exe on WIndows."""
+
+ template_fields: Sequence[str] = "text"
+
+ def __init__(self, text: str, **kwargs):
+ self.text = text
+ super().__init__(**kwargs)
+
+ def execute(self, context: Context) -> Any:
+ tmp_file = Path(gettempdir()) / "airflow_test.txt"
+ with open(tmp_file, "w", encoding="utf8") as textfile:
+ textfile.write(self.text)
+ check_call(["notepad.exe", tmp_file])
+ with open(tmp_file, encoding="utf8") as textfile:
+ return textfile.read()
+
+
+with DAG(
+ dag_id="win_notepad",
+ dag_display_name="Windows Notepad",
+ description=__doc__.partition(".")[0],
+ doc_md=__doc__,
+ schedule=None,
+ start_date=datetime(2024, 7, 1),
+ tags=["edge", "Windows"],
+ default_args={"queue": "windows"},
+ params={
+ "notepad_text": Param(
+ "This is a text as proposal generated by Airflow DAG. Change it and save and it will get to XCom.",
+ title="Notepad Text",
+ description="Add some text that should be filled into Notepad at start.",
+ type="string",
+ format="multiline",
+ ),
+ },
+) as dag:
+ npo = NotepadOperator(
+ task_id="notepad",
+ text="{{ params.notepad_text }}",
+ )
diff --git a/providers/src/airflow/providers/edge/example_dags/win_test.py b/providers/src/airflow/providers/edge/example_dags/win_test.py
new file mode 100644
index 00000000000..b6992ae1f0f
--- /dev/null
+++ b/providers/src/airflow/providers/edge/example_dags/win_test.py
@@ -0,0 +1,339 @@
+#
+# 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.
+"""
+In this DAG some tests are made to check a worker on Windows.
+
+The DAG is created in conjunction with the documentation in
+https://github.com/apache/airflow/blob/main/docs/apache-airflow-providers-edge/install_on_windows.rst
+and serves as a PoC test for the Windows worker.
+"""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Container, Sequence
+from datetime import datetime
+from subprocess import STDOUT, Popen
+from time import sleep
+from typing import TYPE_CHECKING, Any, Callable
+
+from airflow.decorators import task, task_group
+from airflow.exceptions import AirflowException, AirflowNotFoundException, AirflowSkipException
+from airflow.hooks.base import BaseHook
+from airflow.models import BaseOperator
+from airflow.models.dag import DAG
+from airflow.models.param import Param
+from airflow.models.variable import Variable
+from airflow.operators.empty import EmptyOperator
+from airflow.utils.operator_helpers import context_to_airflow_vars
+from airflow.utils.trigger_rule import TriggerRule
+from airflow.utils.types import ArgNotSet
+
+if TYPE_CHECKING:
+ from airflow.models.taskinstance import TaskInstance
+ from airflow.utils.context import Context
+
+try:
+ from airflow.operators.python import PythonOperator # type: ignore
+except ImportError:
+ from airflow.providers.common.compat.standard.operators import PythonOperator
+
+
+class CmdOperator(BaseOperator):
+ r"""Execute a command or batch of commands.
+
+ This operator is forked of BashOperator to execute any process on windows.
+
+ If BaseOperator.do_xcom_push is True, the last line written to stdout
+ will also be pushed to an XCom when the command completes
+
+ :param command: The command, set of commands or reference to a
+ BAT script (must be '.bat') to be executed. (templated)
+ :param env: If env is not None, it must be a dict that defines the
+ environment variables for the new process; these are used instead
+ of inheriting the current process environment, which is the default
+ behavior. (templated)
+ :param append_env: If False(default) uses the environment variables passed in env params
+ and does not inherit the current process environment. If True, inherits the environment variables
+ from current passes and then environment variable passed by the user will either update the existing
+ inherited environment variables or the new variables gets appended to it
+ :param skip_on_exit_code: If task exits with this exit code, leave the task
+ in ``skipped`` state (default: 99). If set to ``None``, any non-zero
+ exit code will be treated as a failure.
+ :param cwd: Working directory to execute the command in (templated).
+ If None (default), the command is run in a temporary directory.
+ To use current DAG folder as the working directory,
+ you might set template ``{{ dag_run.dag.folder }}``.
+ :param output_processor: Function to further process the output of the script / command
+ (default is lambda output: output).
+
+ Airflow will evaluate the exit code of the command. In general, a non-zero exit code will result in
+ task failure and zero will result in task success.
+ Exit code ``99`` (or another set in ``skip_on_exit_code``)
+ will throw an :class:`airflow.exceptions.AirflowSkipException`, which will leave the task in ``skipped``
+ state. You can have all non-zero exit codes be treated as a failure by setting ``skip_on_exit_code=None``.
+
+ .. list-table::
+ :widths: 25 25
+ :header-rows: 1
+
+ * - Exit code
+ - Behavior
+ * - 0
+ - success
+ * - `skip_on_exit_code` (default: 99)
+ - raise :class:`airflow.exceptions.AirflowSkipException`
+ * - otherwise
+ - raise :class:`airflow.exceptions.AirflowException`
+
+ .. warning::
+
+ Care should be taken with "user" input or when using Jinja templates in the
+ ``command``, as this command operator does not perform any escaping or
+ sanitization of the command.
+
+ This applies mostly to using "dag_run" conf, as that can be submitted via
+ users in the Web UI. Most of the default template variables are not at
+ risk.
+
+ """
+
+ template_fields: Sequence[str] = ("command", "env", "cwd")
+ template_fields_renderers = {"command": "bash", "env": "json"}
+ template_ext: Sequence[str] = ".bat"
+
+ subprocess: Popen | None = None
+
+ def __init__(
+ self,
+ *,
+ command: list[str] | str | ArgNotSet,
+ env: dict[str, str] | None = None,
+ append_env: bool = False,
+ skip_on_exit_code: int | Container[int] | None = 99,
+ cwd: str | None = None,
+ output_processor: Callable[[str], Any] = lambda result: result,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.command = command.split(" ") if isinstance(command, str) else command
+ self.env = env
+ self.skip_on_exit_code = (
+ skip_on_exit_code
+ if isinstance(skip_on_exit_code, Container)
+ else [skip_on_exit_code]
+ if skip_on_exit_code is not None
+ else []
+ )
+ self.cwd = cwd
+ self.append_env = append_env
+ self.output_processor = output_processor
+
+ # When using the @task.command decorator, the command is not known until the underlying Python
+ # callable is executed and therefore set to NOTSET initially. This flag is useful during execution to
+ # determine whether the command value needs to re-rendered.
+ self._init_command_not_set = isinstance(self.command, ArgNotSet)
+
+ @staticmethod
+ def refresh_command(ti: TaskInstance) -> None:
+ """Rewrite the underlying rendered command value for a task instance in the metadatabase.
+
+ TaskInstance.get_rendered_template_fields() cannot be used because this will retrieve the
+ RenderedTaskInstanceFields from the metadatabase which doesn't have the runtime-evaluated command
+ value.
+
+ :meta private:
+ """
+ from airflow.models.renderedtifields import RenderedTaskInstanceFields
+
+ RenderedTaskInstanceFields._update_runtime_evaluated_template_fields(ti)
+
+ def get_env(self, context):
+ """Build the set of environment variables to be exposed for the command."""
+ system_env = os.environ.copy()
+ env = self.env
+ if env is None:
+ env = system_env
+ elif self.append_env:
+ system_env.update(env)
+ env = system_env
+
+ airflow_context_vars = context_to_airflow_vars(context, in_env_var_format=True)
+ self.log.debug(
+ "Exporting env vars: %s",
+ " ".join(f"{k}={v!r}" for k, v in airflow_context_vars.items()),
+ )
+ env.update(airflow_context_vars)
+ return env
+
+ def execute(self, context: Context):
+ if self.cwd is not None:
+ if not os.path.exists(self.cwd):
+ raise AirflowException(f"Can not find the cwd: {self.cwd}")
+ if not os.path.isdir(self.cwd):
+ raise AirflowException(f"The cwd {self.cwd} must be a directory")
+ env = self.get_env(context)
+
+ # Because the command value is evaluated at runtime using the @task.command decorator, the
+ # RenderedTaskInstanceField data needs to be rewritten and the command value re-rendered -- the
+ # latter because the returned command from the decorated callable could contain a Jinja expression.
+ # Both will ensure the correct command is executed and that the Rendered Template view in the UI
+ # displays the executed command (otherwise it will display as an ArgNotSet type).
+ if self._init_command_not_set:
+ ti = context["ti"]
+ self.refresh_command(ti)
+
+ self.subprocess = Popen(
+ args=self.command, # type: ignore # here we assume the arg has been replaced by a string array
+ shell=True,
+ env=env,
+ stderr=STDOUT,
+ text=True,
+ cwd=self.cwd,
+ )
+ outs, _ = self.subprocess.communicate()
+ exit_code = self.subprocess.returncode
+ if exit_code in self.skip_on_exit_code:
+ raise AirflowSkipException(f"Command returned exit code {exit_code}. Skipping.")
+ if exit_code != 0:
+ raise AirflowException(f"Command failed. The command returned a non-zero exit code {exit_code}.")
+
+ return self.output_processor(outs)
+
+ def on_kill(self) -> None:
+ if self.subprocess:
+ self.subprocess.kill()
+
+
+with DAG(
+ dag_id="win_test",
+ dag_display_name="Windows Test",
+ description=__doc__.partition(".")[0],
+ doc_md=__doc__,
+ schedule=None,
+ start_date=datetime(2025, 1, 1),
+ tags=["edge", "Windows"],
+ default_args={"queue": "windows"},
+ params={
+ "mapping_count": Param(
+ 4,
+ type="integer",
+ title="Mapping Count",
+ description="Amount of tasks that should be mapped",
+ ),
+ },
+) as dag:
+
+ @task
+ def my_setup():
+ print("Assume this is a setup task")
+
+ @task
+ def mapping_from_params(**context) -> list[int]:
+ mapping_count: int = context["params"]["mapping_count"]
+ return list(range(1, mapping_count + 1))
+
+ @task
+ def add_one(x: int):
+ return x + 1
+
+ @task
+ def sum_it(values):
+ total = sum(values)
+ print(f"Total was {total}")
+
+ @task_group(prefix_group_id=False)
+ def mapping_task_group():
+ added_values = add_one.expand(x=mapping_from_params())
+ sum_it(added_values)
+
+ @task_group(prefix_group_id=False)
+ def standard_tasks_group():
+ @task.branch
+ def branching():
+ return ["virtualenv", "variable", "connection", "command", "classic_python"]
+
+ @task.virtualenv(requirements="numpy")
+ def virtualenv():
+ import numpy # type: ignore
+
+ print(f"Welcome to virtualenv with numpy version {numpy.__version__}.")
+
+ @task
+ def variable():
+ print("Creating a new variable...")
+ Variable.set("integration_test_key", "value")
+ print(f"For the moment the variable is set to {Variable.get('integration_test_key')}")
+ print("Deleting variable...")
+ Variable.delete("integration_test_key")
+
+ @task
+ def connection():
+ try:
+ conn = BaseHook.get_connection("integration_test")
+ print(f"Got connection {conn}")
+ except AirflowNotFoundException:
+ print("Connection 'integration_test' not found... but also OK.")
+
+ command = CmdOperator(task_id="command", command="echo Parameter is {{params.mapping_count}}")
+
+ def python_call():
+ print("Hello world")
+
+ classic_py = PythonOperator(task_id="classic_python", python_callable=python_call)
+
+ empty = EmptyOperator(task_id="not_executed")
+
+ branching() >> [virtualenv(), variable(), connection(), command, classic_py, empty]
+
+ @task_group(prefix_group_id=False)
+ def failure_tests_group():
+ @task
+ def plan_to_fail():
+ print("This task is supposed to fail")
+ raise ValueError("This task is supposed to fail")
+
+ @task(retries=1, retry_delay=5.0)
+ def needs_retry(**context):
+ print("This task is supposed to fail on the first attempt")
+ if context["ti"].try_number == 1:
+ raise ValueError("This task is supposed to fail")
+
+ @task(trigger_rule=TriggerRule.ONE_SUCCESS)
+ def capture_fail():
+ print("all good, we accept the fail and report OK")
+
+ [plan_to_fail(), needs_retry()] >> capture_fail()
+
+ @task
+ def long_running():
+ print("This task runs for 15 minutes")
+ for i in range(15):
+ sleep(60)
+ print(f"Running for {i + 1} minutes now.")
+ print("Long running task completed.")
+
+ @task
+ def my_teardown():
+ print("Assume this is a teardown task")
+
+ (
+ my_setup().as_setup()
+ >> [mapping_task_group(), standard_tasks_group(), failure_tests_group(), long_running()]
+ >> my_teardown().as_teardown()
+ )
diff --git a/scripts/in_container/bin/run_tmux b/scripts/in_container/bin/run_tmux
index ce6bf801dc4..25725b40912 100755
--- a/scripts/in_container/bin/run_tmux
+++ b/scripts/in_container/bin/run_tmux
@@ -101,7 +101,8 @@ if [[ ${AIRFLOW__CORE__EXECUTOR} == "airflow.providers.edge.executors.edge_execu
# Ensure logs are smelling like Remote and are not visible to other components
tmux send-keys 'export AIRFLOW__LOGGING__BASE_LOG_FOLDER=edge_logs' C-m
- tmux send-keys 'airflow edge worker --edge-hostname breeze' C-m
+ # Start Edge Worker and make a "breeze" hostname, let it pick only default queue
+ tmux send-keys 'airflow edge worker --edge-hostname breeze --queues default' C-m
fi
if [[ ${STANDALONE_DAG_PROCESSOR} == "true" ]]; then
tmux select-pane -t 3