kedro-org / kedro-org/kedro-plugins

How to coordinate Airflow run with versioning

Open
#293 0 comments 0 reactions 0 assignees View on GitHub
technical documentation
Dominant language
Python
Stars
119
Forks
136
Avg merge
4d 10h
Merged PRs (30d)
5

Description

## Description
Is your feature request related to a problem? A clear and concise description of what the problem is: "I'm always frustrated when ..."

User need to pass a value of `model_version` to the Kedro pipeline. It works fine locally but when deployed as Airflow DAGs, it is triggered as many indivdiual KedroOperator. There is no easy way to share information across the DAGs.

This issue document a solution reported by an user. Note that this works but if we need to use the build in `versioned: true` feature it won't work because each `KedroSession` has its own `session_id`. https://github.com/kedro-org/kedro/issues/1731

## More Context
The user here want to use current timestamp as `model_version`, which has similar idea to `session_id`.

```python
def create_model_version(**kwargs):
model_version = datetime.now().strftime('%Y%m%d-%H%M%S')
kwargs['ti'].xcom_push(key='model_version', value=model_version)
```

# Workaround
```python
with DAG(
"test-fi",
start_date=datetime(2023, 1, 1),
max_active_runs=3,
schedule_interval=timedelta(days=30),
default_args=default_args,
catchup=False
) as dag:
tasks = {}

def create_model_version(**kwargs):
model_version = datetime.now().strftime('%Y%m%d-%H%M%S')
kwargs['ti'].xcom_push(key='model_version', value=model_version)

def define_project_parameters_task(**kwargs):
ti = kwargs['ti']
model_version = ti.xcom_pull(task_ids='create_model_version', key='model_version')
print(f'Model version {model_version}')
# Use the model_version in your KedroOperator configuration
define_project_parameters_task = KedroOperator(
task_id="define-project-parameters",
package_name=package_name,
pipeline_name=pipeline_name,
node_name="define_project_parameters",
project_path=project_path,
env=env,
params={"model_version": model_version} # Pass the model_version as a parameter
)
```

Contributor guide

Open the contributing guide

Research direction

Start with the Airflow DAG workaround in the issue, especially create_model_version, XCom, and the KedroOperator configuration that receives model_version. Review how separate KedroOperator tasks share run metadata and how the built-in versioned feature relates to session_id; done should define a supported way to coordinate the value across the DAG.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
data-engineering, devops
Issue type
Feature
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.