annotating-task-lineage

Name: annotating-task-lineage
Author: astronomer

802installs55forks

How to install annotating-task-lineage?

astronomer/agents/annotating-task-lineage

$npx -y skills add astronomer/agents --skill annotating-task-lineage

Installs into the current project.

Summary

Annotate Airflow tasks with data lineage using inlets and outlets. Use when the user wants to add lineage metadata to tasks, specify input/output datasets, or enable lineage tracking for operators without built-in OpenLineage extraction.

Files · 1

View on GitHub

SKILL.md

1	# Annotating Task Lineage with Inlets & Outlets
2
3	This skill guides you through adding manual lineage annotations to Airflow tasks using `inlets` and `outlets`.
4
5	> Reference: See the [OpenLineage provider developer guide](https://airflow.apache.org/docs/apache-airflow-providers-openlineage/stable/guides/developer.html) for the latest supported operators and patterns.
6
7	### On Astro
8
9	Lineage annotations defined with inlets and outlets are visualized in Astro's enhanced Lineage tab, which provides cross-DAG and cross-deployment lineage views. This means your annotations are immediately visible in the Astro UI, giving you a unified view of data flow across your entire Astro organization.
10
11	## When to Use This Approach
12
13	\| Scenario \| Use Inlets/Outlets? \|
14	\|----------\|---------------------\|
15	\| Operator has OpenLineage methods (`get_openlineage_facets_on_*`) \| ❌ Modify the OL method directly \|
16	\| Operator has no built-in OpenLineage extractor \| ✅ Yes \|
17	\| Simple table-level lineage is sufficient \| ✅ Yes \|
18	\| Quick lineage setup without custom code \| ✅ Yes \|
19	\| Need column-level lineage \| ❌ Use OpenLineage methods or custom extractor \|
20	\| Complex extraction logic needed \| ❌ Use OpenLineage methods or custom extractor \|
21
22	> Note: Inlets/outlets are the lowest-priority fallback. If an OpenLineage extractor or method exists for the operator, it takes precedence. Use this approach for operators without extractors.
23
24	---
25
26	## Supported Types for Inlets/Outlets
27
28	You can use OpenLineage Dataset objects or Airflow Assets for inlets and outlets:
29
30	### OpenLineage Datasets (Recommended)
31
32	```python
33	from openlineage.client.event_v2 import Dataset
34
35	# Database tables
36	source_table = Dataset(
37	namespace="postgres://mydb:5432",
38	name="public.orders",
39	)
40	target_table = Dataset(
41	namespace="snowflake://account.snowflakecomputing.com",
42	name="staging.orders_clean",
43	)
44
45	# Files
46	input_file = Dataset(
47	namespace="s3://my-bucket",
48	name="raw/events/2024-01-01.json",
49	)
50	```
51
52	### Airflow Assets (Airflow 3+)
53
54	```python
55	from airflow.sdk import Asset
56
57	# Using Airflow's native Asset type
58	orders_asset = Asset(uri="s3://my-bucket/data/orders")
59	```
60
61	### Airflow Datasets (Airflow 2.4+)
62
63	```python
64	from airflow.datasets import Dataset
65
66	# Using Airflow's Dataset type (Airflow 2.4-2.x)
67	orders_dataset = Dataset(uri="s3://my-bucket/data/orders")
68	```
69
70	---
71
72	## Basic Usage
73
74	### Setting Inlets and Outlets on Operators
75
76	```python
77	from airflow import DAG
78	from airflow.operators.bash import BashOperator
79	from openlineage.client.event_v2 import Dataset
80	import pendulum
81
82	# Define your lineage datasets
83	source_table = Dataset(
84	namespace="snowflake://account.snowflakecomputing.com",
85	name="raw.orders",
86	)
87	target_table = Dataset(
88	namespace="snowflake://account.snowflakecomputing.com",
89	name="staging.orders_clean",
90	)
91	output_file = Dataset(
92	namespace="s3://my-bucket",
93	name="exports/orders.parquet",
94	)
95
96	with DAG(
97	dag_id="etl_with_lineage",
98	start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
99	schedule="@daily",
100	) as dag:
101
102	transform = BashOperator(
103	task_id="transform_orders",
104	bash_command="echo 'transforming...'",
105	inlets=[source_table], # What this task reads
106	outlets=[target_table], # What this task writes
107	)
108
109	export = BashOperator(
110	task_id="export_to_s3",
111	bash_command="echo 'exporting...'",
112	inlets=[target_table], # Reads from previous output
113	outlets=[output_file], # Writes to S3
114	)
115
116	transform >> export
117	```
118
119	### Multiple Inputs and Outputs
120
121	Tasks often read from multiple sources and write to multiple destinations:
122
123	```python
124	from openlineage.client.event_v2 import Dataset
125
126	# Multiple source tables
127	customers = Dataset(namespace="postgres://crm:5432", name="public.customers")
128	orders = Dataset(namespace="postgres://sales:5432", name="public.orders")
129	products = Dataset(namespace="postgres://inventory:5432", name="public.products")
130
131	# Multiple output tables
132	daily_summary = Dataset(namespace="snowflake://account", name="analytics.daily_summary")
133	customer_metrics = Dataset(namespace="snowflake://account", name="analytics.customer_metrics")
134
135	aggregate_task = PythonOperator(
136	task_id="build_daily_aggregates",
137	python_callable=build_aggregates,
138	inlets=[customers, orders, products], # All inputs
139	outlets=[daily_summary, customer_metrics], # All outputs
140	)
141	```
142
143	---
144
145	## Setting Lineage in Custom Operators
146
147	When building custom operators, you have two options:
148
149	### Option 1: Implement OpenLineage Methods (Recommended)
150
151	This is the preferred approach as it gives you full control over lineage extraction:
152
153	```python
154	fro

People who liked this also viewed

6 picks

Type

annotating-task-lineage

802installs55forks

View on GitHub

How to install annotating-task-lineage?

astronomer/agents/annotating-task-lineage

$npx -y skills add astronomer/agents --skill annotating-task-lineage

Installs into the current project.

Summary

Files · 1

View on GitHub

SKILL.md

1	# Annotating Task Lineage with Inlets & Outlets
2
3	This skill guides you through adding manual lineage annotations to Airflow tasks using `inlets` and `outlets`.
4
5	> Reference: See the [OpenLineage provider developer guide](https://airflow.apache.org/docs/apache-airflow-providers-openlineage/stable/guides/developer.html) for the latest supported operators and patterns.
6
7	### On Astro
8
9	Lineage annotations defined with inlets and outlets are visualized in Astro's enhanced Lineage tab, which provides cross-DAG and cross-deployment lineage views. This means your annotations are immediately visible in the Astro UI, giving you a unified view of data flow across your entire Astro organization.
10
11	## When to Use This Approach
12
13	\| Scenario \| Use Inlets/Outlets? \|
14	\|----------\|---------------------\|
15	\| Operator has OpenLineage methods (`get_openlineage_facets_on_*`) \| ❌ Modify the OL method directly \|
16	\| Operator has no built-in OpenLineage extractor \| ✅ Yes \|
17	\| Simple table-level lineage is sufficient \| ✅ Yes \|
18	\| Quick lineage setup without custom code \| ✅ Yes \|
19	\| Need column-level lineage \| ❌ Use OpenLineage methods or custom extractor \|
20	\| Complex extraction logic needed \| ❌ Use OpenLineage methods or custom extractor \|
21
22	> Note: Inlets/outlets are the lowest-priority fallback. If an OpenLineage extractor or method exists for the operator, it takes precedence. Use this approach for operators without extractors.
23
24	---
25
26	## Supported Types for Inlets/Outlets
27
28	You can use OpenLineage Dataset objects or Airflow Assets for inlets and outlets:
29
30	### OpenLineage Datasets (Recommended)
31
32	```python
33	from openlineage.client.event_v2 import Dataset
34
35	# Database tables
36	source_table = Dataset(
37	namespace="postgres://mydb:5432",
38	name="public.orders",
39	)
40	target_table = Dataset(
41	namespace="snowflake://account.snowflakecomputing.com",
42	name="staging.orders_clean",
43	)
44
45	# Files
46	input_file = Dataset(
47	namespace="s3://my-bucket",
48	name="raw/events/2024-01-01.json",
49	)
50	```
51
52	### Airflow Assets (Airflow 3+)
53
54	```python
55	from airflow.sdk import Asset
56
57	# Using Airflow's native Asset type
58	orders_asset = Asset(uri="s3://my-bucket/data/orders")
59	```
60
61	### Airflow Datasets (Airflow 2.4+)
62
63	```python
64	from airflow.datasets import Dataset
65
66	# Using Airflow's Dataset type (Airflow 2.4-2.x)
67	orders_dataset = Dataset(uri="s3://my-bucket/data/orders")
68	```
69
70	---
71
72	## Basic Usage
73
74	### Setting Inlets and Outlets on Operators
75
76	```python
77	from airflow import DAG
78	from airflow.operators.bash import BashOperator
79	from openlineage.client.event_v2 import Dataset
80	import pendulum
81
82	# Define your lineage datasets
83	source_table = Dataset(
84	namespace="snowflake://account.snowflakecomputing.com",
85	name="raw.orders",
86	)
87	target_table = Dataset(
88	namespace="snowflake://account.snowflakecomputing.com",
89	name="staging.orders_clean",
90	)
91	output_file = Dataset(
92	namespace="s3://my-bucket",
93	name="exports/orders.parquet",
94	)
95
96	with DAG(
97	dag_id="etl_with_lineage",
98	start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
99	schedule="@daily",
100	) as dag:
101
102	transform = BashOperator(
103	task_id="transform_orders",
104	bash_command="echo 'transforming...'",
105	inlets=[source_table], # What this task reads
106	outlets=[target_table], # What this task writes
107	)
108
109	export = BashOperator(
110	task_id="export_to_s3",
111	bash_command="echo 'exporting...'",
112	inlets=[target_table], # Reads from previous output
113	outlets=[output_file], # Writes to S3
114	)
115
116	transform >> export
117	```
118
119	### Multiple Inputs and Outputs
120
121	Tasks often read from multiple sources and write to multiple destinations:
122
123	```python
124	from openlineage.client.event_v2 import Dataset
125
126	# Multiple source tables
127	customers = Dataset(namespace="postgres://crm:5432", name="public.customers")
128	orders = Dataset(namespace="postgres://sales:5432", name="public.orders")
129	products = Dataset(namespace="postgres://inventory:5432", name="public.products")
130
131	# Multiple output tables
132	daily_summary = Dataset(namespace="snowflake://account", name="analytics.daily_summary")
133	customer_metrics = Dataset(namespace="snowflake://account", name="analytics.customer_metrics")
134
135	aggregate_task = PythonOperator(
136	task_id="build_daily_aggregates",
137	python_callable=build_aggregates,
138	inlets=[customers, orders, products], # All inputs
139	outlets=[daily_summary, customer_metrics], # All outputs
140	)
141	```
142
143	---
144
145	## Setting Lineage in Custom Operators
146
147	When building custom operators, you have two options:
148
149	### Option 1: Implement OpenLineage Methods (Recommended)
150
151	This is the preferred approach as it gives you full control over lineage extraction:
152
153	```python
154	fro

People who liked this also viewed

6 picks

Type