kacpermuda commented on code in PR #65727:
URL: https://github.com/apache/airflow/pull/65727#discussion_r3160832408
##########
providers/openlineage/src/airflow/providers/openlineage/extractors/manager.py:
##########
@@ -174,18 +175,30 @@ def _get_extractor(self, task: BaseOperator) ->
BaseExtractor | None:
return extractor(task)
return None
- def extract_inlets_and_outlets(self, task_metadata: OperatorLineage, task)
-> None:
+ def extract_inlets_and_outlets(
+ self,
+ task_metadata: OperatorLineage,
+ task,
+ task_instance=None,
+ ) -> None:
if task.inlets or task.outlets:
self.log.debug("Manually extracting lineage metadata from inlets
and outlets")
for i in task.inlets:
- d = self.convert_to_ol_dataset(i)
- if d:
+ if d := self.convert_to_ol_dataset(i):
task_metadata.inputs.append(d)
for o in task.outlets:
- d = self.convert_to_ol_dataset(o)
- if d:
+ if d := self.convert_to_ol_dataset(o):
task_metadata.outputs.append(d)
+ # Add runtime-emitted outlets (alias resolutions + dynamic asset
events), deduped
+ # by (namespace, name) against both static outputs and each other.
+ seen = {(d.namespace, d.name) for d in task_metadata.outputs}
+ for asset, _alias in get_runtime_outlet_assets(task_instance):
+ ol = translate_airflow_asset(asset, None)
Review Comment:
That makes sense. Tried to do this quickly now, but it requires some changes
to be detailed enough (don't want to just log number of dropped ones, we can do
better than that). Will create a followup ticket to improve logging here.
Thanks !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
