kedro-org / kedro-org/kedro-plugins
How to coordinate Airflow run with versioning
- 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
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