creating-openlineage-extractors

Name: creating-openlineage-extractors
Author: astronomer

779installs55forks

How to install creating-openlineage-extractors?

astronomer/agents/creating-openlineage-extractors

$npx -y skills add astronomer/agents --skill creating-openlineage-extractors

Installs into the current project.

Summary

Create custom OpenLineage extractors for Airflow operators. Use when the user needs lineage from unsupported or third-party operators, wants column-level lineage, or needs complex extraction logic beyond what inlets/outlets provide.

Files · 1

View on GitHub

SKILL.md

1	# Creating OpenLineage Extractors
2
3	This skill guides you through creating custom OpenLineage extractors to capture lineage from Airflow operators that don't have built-in support.
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 patterns and list of supported operators/hooks.
6
7	## When to Use Each Approach
8
9	\| Scenario \| Approach \|
10	\|----------\|----------\|
11	\| Operator you own/maintain \| OpenLineage Methods (recommended, simplest) \|
12	\| Third-party operator you can't modify \| Custom Extractor \|
13	\| Need column-level lineage \| OpenLineage Methods or Custom Extractor \|
14	\| Complex extraction logic \| OpenLineage Methods or Custom Extractor \|
15	\| Simple table-level lineage \| Inlets/Outlets (simplest, but lowest priority) \|
16
17	> Important: Always prefer OpenLineage methods over custom extractors when possible. Extractors are harder to write, easier to diverge from operator behavior after changes, and harder to debug.
18
19	### On Astro
20
21	Astro includes built-in OpenLineage integration — no additional transport configuration is needed. Lineage events are automatically collected and displayed in the Astro UI's Lineage tab. Custom extractors deployed to an Astro project are automatically picked up, so you only need to register them in `airflow.cfg` or via environment variable and deploy.
22
23	---
24
25	## Two Approaches
26
27	### 1. OpenLineage Methods (Recommended)
28
29	Use when you can add methods directly to your custom operator. This is the go-to solution for operators you own.
30
31	### 2. Custom Extractors
32
33	Use when you need lineage from third-party or provider operators that you cannot modify.
34
35	---
36
37	## Approach 1: OpenLineage Methods (Recommended)
38
39	When you own the operator, add OpenLineage methods directly:
40
41	```python
42	from airflow.models import BaseOperator
43
44
45	class MyCustomOperator(BaseOperator):
46	"""Custom operator with built-in OpenLineage support."""
47
48	def __init__(self, source_table: str, target_table: str, **kwargs):
49	super().__init__(**kwargs)
50	self.source_table = source_table
51	self.target_table = target_table
52	self._rows_processed = 0 # Set during execution
53
54	def execute(self, context):
55	# Do the actual work
56	self._rows_processed = self._process_data()
57	return self._rows_processed
58
59	def get_openlineage_facets_on_start(self):
60	"""Called when task starts. Return known inputs/outputs."""
61	# Import locally to avoid circular imports
62	from openlineage.client.event_v2 import Dataset
63	from airflow.providers.openlineage.extractors import OperatorLineage
64
65	return OperatorLineage(
66	inputs=[Dataset(namespace="postgres://db", name=self.source_table)],
67	outputs=[Dataset(namespace="postgres://db", name=self.target_table)],
68	)
69
70	def get_openlineage_facets_on_complete(self, task_instance):
71	"""Called after success. Add runtime metadata."""
72	from openlineage.client.event_v2 import Dataset
73	from openlineage.client.facet_v2 import output_statistics_output_dataset
74	from airflow.providers.openlineage.extractors import OperatorLineage
75
76	return OperatorLineage(
77	inputs=[Dataset(namespace="postgres://db", name=self.source_table)],
78	outputs=[
79	Dataset(
80	namespace="postgres://db",
81	name=self.target_table,
82	facets={
83	"outputStatistics": output_statistics_output_dataset.OutputStatisticsOutputDatasetFacet(
84	rowCount=self._rows_processed
85	)
86	},
87	)
88	],
89	)
90
91	def get_openlineage_facets_on_failure(self, task_instance):
92	"""Called after failure. Optional - for partial lineage."""
93	return None
94	```
95
96	### OpenLineage Methods Reference
97
98	\| Method \| When Called \| Required \|
99	\|--------\|-------------\|----------\|
100	\| `get_openlineage_facets_on_start()` \| Task enters RUNNING \| No \|
101	\| `get_openlineage_facets_on_complete(ti)` \| Task succeeds \| No \|
102	\| `get_openlineage_facets_on_failure(ti)` \| Task fails \| No \|
103
104	> Implement only the methods you need. Unimplemented methods fall through to Hook-Level Lineage or inlets/outlets.
105
106	---
107
108	## Approach 2: Custom Extractors
109
110	Use this approach only when you cannot modify the operator (e.g., third-party or provider operators).
111
112	### Basic Structure
113
114	```python
115	from airflow.providers.openlineage.extractors.base import BaseExtractor, OperatorLineage
116	from openlineage.client.event_v2 import Dataset
117
118
119	class MyOperatorExtractor(BaseExtr

People who liked this also viewed

6 picks

Type

creating-openlineage-extractors

779installs55forks

View on GitHub

How to install creating-openlineage-extractors?

astronomer/agents/creating-openlineage-extractors

$npx -y skills add astronomer/agents --skill creating-openlineage-extractors

Installs into the current project.

Summary

Files · 1

View on GitHub

SKILL.md

1	# Creating OpenLineage Extractors
2
3	This skill guides you through creating custom OpenLineage extractors to capture lineage from Airflow operators that don't have built-in support.
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 patterns and list of supported operators/hooks.
6
7	## When to Use Each Approach
8
9	\| Scenario \| Approach \|
10	\|----------\|----------\|
11	\| Operator you own/maintain \| OpenLineage Methods (recommended, simplest) \|
12	\| Third-party operator you can't modify \| Custom Extractor \|
13	\| Need column-level lineage \| OpenLineage Methods or Custom Extractor \|
14	\| Complex extraction logic \| OpenLineage Methods or Custom Extractor \|
15	\| Simple table-level lineage \| Inlets/Outlets (simplest, but lowest priority) \|
16
17	> Important: Always prefer OpenLineage methods over custom extractors when possible. Extractors are harder to write, easier to diverge from operator behavior after changes, and harder to debug.
18
19	### On Astro
20
21	Astro includes built-in OpenLineage integration — no additional transport configuration is needed. Lineage events are automatically collected and displayed in the Astro UI's Lineage tab. Custom extractors deployed to an Astro project are automatically picked up, so you only need to register them in `airflow.cfg` or via environment variable and deploy.
22
23	---
24
25	## Two Approaches
26
27	### 1. OpenLineage Methods (Recommended)
28
29	Use when you can add methods directly to your custom operator. This is the go-to solution for operators you own.
30
31	### 2. Custom Extractors
32
33	Use when you need lineage from third-party or provider operators that you cannot modify.
34
35	---
36
37	## Approach 1: OpenLineage Methods (Recommended)
38
39	When you own the operator, add OpenLineage methods directly:
40
41	```python
42	from airflow.models import BaseOperator
43
44
45	class MyCustomOperator(BaseOperator):
46	"""Custom operator with built-in OpenLineage support."""
47
48	def __init__(self, source_table: str, target_table: str, **kwargs):
49	super().__init__(**kwargs)
50	self.source_table = source_table
51	self.target_table = target_table
52	self._rows_processed = 0 # Set during execution
53
54	def execute(self, context):
55	# Do the actual work
56	self._rows_processed = self._process_data()
57	return self._rows_processed
58
59	def get_openlineage_facets_on_start(self):
60	"""Called when task starts. Return known inputs/outputs."""
61	# Import locally to avoid circular imports
62	from openlineage.client.event_v2 import Dataset
63	from airflow.providers.openlineage.extractors import OperatorLineage
64
65	return OperatorLineage(
66	inputs=[Dataset(namespace="postgres://db", name=self.source_table)],
67	outputs=[Dataset(namespace="postgres://db", name=self.target_table)],
68	)
69
70	def get_openlineage_facets_on_complete(self, task_instance):
71	"""Called after success. Add runtime metadata."""
72	from openlineage.client.event_v2 import Dataset
73	from openlineage.client.facet_v2 import output_statistics_output_dataset
74	from airflow.providers.openlineage.extractors import OperatorLineage
75
76	return OperatorLineage(
77	inputs=[Dataset(namespace="postgres://db", name=self.source_table)],
78	outputs=[
79	Dataset(
80	namespace="postgres://db",
81	name=self.target_table,
82	facets={
83	"outputStatistics": output_statistics_output_dataset.OutputStatisticsOutputDatasetFacet(
84	rowCount=self._rows_processed
85	)
86	},
87	)
88	],
89	)
90
91	def get_openlineage_facets_on_failure(self, task_instance):
92	"""Called after failure. Optional - for partial lineage."""
93	return None
94	```
95
96	### OpenLineage Methods Reference
97
98	\| Method \| When Called \| Required \|
99	\|--------\|-------------\|----------\|
100	\| `get_openlineage_facets_on_start()` \| Task enters RUNNING \| No \|
101	\| `get_openlineage_facets_on_complete(ti)` \| Task succeeds \| No \|
102	\| `get_openlineage_facets_on_failure(ti)` \| Task fails \| No \|
103
104	> Implement only the methods you need. Unimplemented methods fall through to Hook-Level Lineage or inlets/outlets.
105
106	---
107
108	## Approach 2: Custom Extractors
109
110	Use this approach only when you cannot modify the operator (e.g., third-party or provider operators).
111
112	### Basic Structure
113
114	```python
115	from airflow.providers.openlineage.extractors.base import BaseExtractor, OperatorLineage
116	from openlineage.client.event_v2 import Dataset
117
118
119	class MyOperatorExtractor(BaseExtr

People who liked this also viewed

6 picks

Type