Metadata-Version: 2.4
Name: kedro-airflow
Version: 0.11.0
Summary: Kedro-Airflow makes it easy to deploy Kedro projects to Airflow
Author: Kedro
License: Apache Software License (Apache 2.0)
Project-URL: Source, https://github.com/kedro-org/kedro-plugins/tree/main/kedro-airflow
Project-URL: Documentation, https://github.com/kedro-org/kedro-plugins/blob/main/kedro-airflow/README.md
Project-URL: Tracker, https://github.com/kedro-org/kedro-plugins/issues
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: kedro<2.0.0,>=1.0.0rc1
Requires-Dist: python-slugify>=4.0
Requires-Dist: semver>=2.10
Provides-Extra: test
Requires-Dist: apache-airflow<4.0,>=3.2.0; extra == "test"
Requires-Dist: behave; extra == "test"
Requires-Dist: coverage>=7.2.0; extra == "test"
Requires-Dist: kedro-datasets; extra == "test"
Requires-Dist: pytest; extra == "test"
Requires-Dist: pytest-cov; extra == "test"
Requires-Dist: pytest-mock; extra == "test"
Requires-Dist: pytest-xdist; extra == "test"
Requires-Dist: wheel; extra == "test"
Provides-Extra: lint
Requires-Dist: bandit; extra == "lint"
Requires-Dist: black~=24.0; extra == "lint"
Requires-Dist: detect-secrets~=1.5.0; extra == "lint"
Requires-Dist: mypy~=1.0; extra == "lint"
Requires-Dist: pre-commit>=2.9.2; extra == "lint"
Requires-Dist: ruff~=0.14.4; extra == "lint"
Requires-Dist: types-PyYAML; extra == "lint"
Requires-Dist: types-cachetools; extra == "lint"
Requires-Dist: types-toml; extra == "lint"

# Kedro-Airflow

[![License](https://img.shields.io/badge/license-Apache%202.0-blue.svg)](https://opensource.org/licenses/Apache-2.0)
[![Python Version](https://img.shields.io/badge/python-3.10%20%7C%203.11%20%7C%203.12%20%7C%203.13%20%7C%203.14-blue.svg)](https://pypi.org/project/kedro-airflow/)
[![PyPI Version](https://badge.fury.io/py/kedro-airflow.svg)](https://pypi.org/project/kedro-airflow/)
[![Code Style: Black](https://img.shields.io/badge/code%20style-black-black.svg)](https://github.com/ambv/black)

[Apache Airflow](https://github.com/apache/airflow) is a tool for orchestrating complex workflows and data processing pipelines. The Kedro-Airflow plugin can be used for:
- Rapid pipeline creation in the prototyping phase. You can write Python functions in Kedro without worrying about schedulers, daemons, services or having to recreate the Airflow DAG file.
- Automatic dependency resolution in Kedro. This allows you to bypass Airflow's need to specify the order of your tasks.
- Distributing Kedro tasks across many workers. You can also enable monitoring and scheduling of the tasks' runtimes.

## Installation

`kedro-airflow` is a Python plugin. To install it:

```bash
pip install kedro-airflow
```

## Usage

You can use `kedro-airflow` to deploy a Kedro pipeline as an Airflow DAG by following these steps:

### Step 1: Generate the DAG file

At the root directory of the Kedro project, run:

```bash
kedro airflow create
```

This command will generate an Airflow DAG file located in the `airflow_dags/` directory in your project.
You can pass the `--pipelines` flag to generate the DAG file for a specific Kedro pipeline and the `--env` flag to generate the DAG file for a specific Kedro environment.
Passing `--all` will convert all registered Kedro pipelines to Airflow DAGs.

### Step 2: Copy the DAG file to the Airflow DAGs folder.

For more information about the DAGs folder, please visit [Airflow documentation](https://airflow.apache.org/docs/stable/concepts.html#dags).
The Airflow DAG configuration can be customized by editing this file.

### Step 3: Package and install the Kedro pipeline in the Airflow executor's environment

After generating and deploying the DAG file, you will then need to package and install the Kedro pipeline into the Airflow executor's environment.
Please visit the guide to [Apache Airflow deployment](https://docs.kedro.org/en/stable/deployment/airflow.html) for more details.

### FAQ

#### What if my DAG file is in a different directory to my project folder?

By default, the generated DAG file is configured to live in the same directory as your project as per this [template](https://github.com/kedro-org/kedro-plugins/blob/main/kedro-airflow/kedro_airflow/airflow_dag_template.j2#L44). If your DAG file is located in a different directory to your project, you will need to tweak this  manually after running the `kedro airflow create` command.

#### What if I want to use a different Jinja2 template?

You can use the additional command line argument `--jinja-file` (alias `-j`) to provide an alternative path to a Jinja2 template. Note that these files have to accept the same variables as those used in the [default Jinja2 template](https://github.com/kedro-org/kedro-plugins/blob/main/kedro-airflow/kedro_airflow/airflow_dag_template.j2).

```bash
kedro airflow create --jinja-file=./custom/template.j2
```

#### How can I pass arguments to the Airflow DAGs dynamically?

`kedro-airflow` picks up configuration from `airflow.yml` in `conf/base` or `conf/local` of your Kedro project.
Or it could be in a folder starting with `airflow`.
The [parameters](https://docs.kedro.org/en/stable/configuration/parameters.html) are read by Kedro.
Arguments can be specified globally, or per pipeline:

```yaml
# Global parameters
default:
    start_date: [2023, 1, 1]
    max_active_runs: 3
    # https://airflow.apache.org/docs/stable/scheduler.html#dag-runs
    schedule_interval: "@once"
    catchup: false
    # Default settings applied to all tasks
    owner: "airflow"
    depends_on_past: false
    email_on_failure: false
    email_on_retry: false
    retries: 1
    retry_delay: 5

# Arguments specific to the pipeline (overrides the parameters above)
data_science:
    owner: "airflow-ds"
```

Arguments can also be passed via `--params` in the command line:

```bash
kedro airflow create --params "schedule_interval='@weekly'"
```

These variables are passed to the Jinja2 template that creates an Airflow DAG from your pipeline.

### What if I want to use a configuration pattern other than `airflow*` and `airflow**`?

In order to configure the config loader, update the `settings.py` file in your Kedro project.
For instance, if you would like to use the name `scheduler`, then change the file as follows:

```python
CONFIG_LOADER_ARGS = {"config_patterns": {"airflow": ["scheduler*", "scheduler/**"]}}
```

Follow Kedro's [official documentation](https://docs.kedro.org/en/stable/configuration/advanced_configuration.html#how-to-do-templating-with-the-omegaconfigloader), to see how to add templating, custom resolvers etc.

#### What if I want to pass different arguments?

In order to pass arguments other than those specified in the default template, simply pass a custom template (see: _"What if I want to use a different Jinja2 template?"_)

The syntax for arguments is:
```
{{ argument_name }}
```

In order to make arguments optional, one can use:
```
{{ argument_name | default("default_value") }}
```

For examples, please have a look at the default template (`airflow_dag_template.j2`).

### What if I want to use a configuration file other than `airflow.yml`?

The default configuration pattern is `["airflow*", "airflow/**"]`.
In order to configure the `OmegaConfigLoader`, update the `settings.py` file in your Kedro project as follows:

```python
from kedro.config import OmegaConfigLoader

CONFIG_LOADER_CLASS = OmegaConfigLoader
CONFIG_LOADER_ARGS = {
    # other args
    "config_patterns": {  # configure the pattern for configuration files
        "airflow": ["airflow*", "airflow/**"]
    }
}
```

Follow Kedro's official documentation, to see how to add templating, custom resolvers etc. (https://docs.kedro.org/en/stable/configuration/advanced_configuration.html#how-to-do-templating-with-the-omegaconfigloader)[https://docs.kedro.org/en/stable/configuration/advanced_configuration.html#how-to-do-templating-with-the-omegaconfigloader]

#### How can I use Airflow runtime parameters?

It is possible to pass parameters when triggering an Airflow DAG from the user interface.
In order to use this feature, create a custom template using the [Params syntax](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/params.html).
See ["What if I want to use a different Jinja2 template?"](#what-if-i-want-to-use-a-different-jinja2-template) for instructions on using custom templates.

#### What if I want to use a different Airflow Operator?

Which Airflow Operator to use depends on the environment your project is running in.
You can set the operator to use by providing a custom template.
See ["What if I want to use a different Jinja2 template?"](#what-if-i-want-to-use-a-different-jinja2-template) for instructions on using custom templates.
The [rich offering](https://airflow.apache.org/docs/apache-airflow-providers/operators-and-hooks-ref/index.html) of operators means that the `kedro-airflow` plugin is providing templates for specific operators.
The default template provided by `kedro-airflow` uses the [TaskFlow API](https://airflow.apache.org/docs/apache-airflow/stable/tutorial/taskflow.html) (`@dag` and `@task` decorators from `airflow.sdk`).

### Can I group nodes together?

The `--group-by` option allows you to group Kedro nodes into single Airflow tasks. This is useful for reducing the overhead of task scheduling and for handling datasets that cannot be shared across distributed workers.

#### Grouping by memory

When running Kedro nodes using Airflow, MemoryDatasets are often not shared across operators, which can cause the DAG run to fail.

`MemoryDataset`s may be used to provide logical separation between nodes in Kedro, without the overhead of needing to write to disk (and in the case of distributed running, needing multiple executors).

Nodes that are connected through `MemoryDataset`s can be grouped together using the `--group-by memory` flag:

```bash
kedro airflow create --group-by memory
```

This preserves the option to have logical separation in Kedro, with little computational overhead. Nodes connected via `MemoryDataset`s will be combined into a single Airflow task.

#### Grouping by namespace

If your Kedro pipeline uses [namespaces](https://docs.kedro.org/en/stable/build/namespaces/) to organise nodes, you can group all nodes within the same namespace into a single Airflow task using the `--group-by namespace` flag:

```bash
kedro airflow create --group-by namespace
```

This is particularly useful when:
- You have logically grouped nodes using namespaces and want to execute them together
- You want to reduce the number of Airflow tasks while maintaining the namespace structure from your Kedro pipeline
- Your namespaced nodes share intermediate data that doesn't need to be persisted between tasks

Nodes without a namespace will each be converted to individual Airflow tasks.

For more information about namespaces in Kedro, see the [namespaces documentation](https://docs.kedro.org/en/stable/build/namespaces/).

#### Using task groups for visualisation

It is possible to use [Airflow task groups](https://docs.astronomer.io/learn/task-groups) by changing the template.
See ["What if I want to use a different Jinja2 template?"](#what-if-i-want-to-use-a-different-jinja2-template) for instructions on using custom templates.

## Migrating from Airflow 2.x

`kedro-airflow` **0.11.0 and above** requires **Apache Airflow 3.x** (`>=3.2.0`). The default DAG template was rewritten to use the [TaskFlow API](https://airflow.apache.org/docs/apache-airflow/stable/tutorial/taskflow.html) and is not compatible with Airflow 2.x. If you need Airflow 2.x support, pin to `kedro-airflow<0.11.0`.

### What changed in the generated DAG

The generated DAG no longer defines a `KedroOperator` class. Instead, a shared `_run_kedro_node` helper is defined once and each Kedro node group becomes a `@task`-decorated function that calls it:

```python
# Airflow 3.x (new)
from airflow.sdk import dag, task

def _run_kedro_node(node_names=None, namespaces=None):
    configure_project(package_name)
    with KedroSession.create(project_path=project_path, env=env, conf_source=conf_source) as session:
        if namespaces is not None:
            session.run(pipeline_name=pipeline_name, namespaces=namespaces)
        else:
            session.run(pipeline_name=pipeline_name, node_names=node_names)

@dag(dag_id="my_project", schedule="@once", ...)
def my_project():
    @task(task_id="split")
    def split():
        _run_kedro_node(node_names=["split"])
    ...

my_project()
```

### `--pipeline` renamed to `--pipelines`

The CLI flag `--pipeline` (`-p`) has been renamed to `--pipelines`. It now accepts a comma-separated list of pipeline names, so you can convert multiple pipelines in one invocation:

```bash
kedro airflow create --pipelines data_processing,data_science
```

Update any scripts or CI commands that use `kedro airflow create --pipeline <name>` to use `--pipelines` instead.

### `schedule_interval` in `airflow.yml`

The `schedule_interval` config key in `airflow.yml` continues to work without any changes — it is mapped to the `schedule` parameter in the generated Python code automatically. You can also use `schedule` directly. No migration of your config files is needed.

### Custom templates

If you maintain a custom Jinja2 template via `--jinja-file` that subclasses or references `KedroOperator`, you will need to rewrite it. The recommended approach is to define a `_run_kedro_node` helper and `@task`-decorated functions inside a `@dag` function, mirroring the new default template. All template variables passed by `kedro airflow create` (e.g. `node_objs`, `dag_name`, `env`) remain unchanged.

## Can I contribute?

Yes! Want to help build Kedro-Airflow? Check out our guide to [contributing](https://github.com/kedro-org/kedro-plugins/blob/main/kedro-airflow/CONTRIBUTING.md).

## What licence do you use?

Kedro-Airflow is licensed under the [Apache 2.0](https://github.com/kedro-org/kedro-plugins/blob/main/LICENSE.md) License.

## Python version support policy
* The [Kedro-Airflow](https://github.com/kedro-org/kedro-plugins/tree/main/kedro-airflow) supports all Python versions that are actively maintained by the CPython core team. When a [Python version reaches end of life](https://devguide.python.org/versions/#versions), support for that version is dropped from `kedro-airflow`. This is not considered a breaking change.
