feat(systemMetadata): add pipeline names to system metadata

Follow up on #8672
datahub-project · Aug 21, 2023 · df796d3 · df796d3
1 parent 8cf299a
commit df796d3
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 0 deletions.
diff --git a/metadata-ingestion/src/datahub/ingestion/extractor/mce_extractor.py b/metadata-ingestion/src/datahub/ingestion/extractor/mce_extractor.py
@@ -27,6 +27,9 @@ def _try_reformat_with_black(code: str) -> str:
 
 class WorkUnitRecordExtractorConfig(ConfigModel):
     set_system_metadata = True
+    set_system_metadata_pipeline_name = (
+        False  # false for now until the models are available in OSS
+    )
     unpack_mces_into_mcps = False
 
 
@@ -66,6 +69,10 @@ def get_records(
                     workunit.metadata.systemMetadata = SystemMetadata(
                         lastObserved=get_sys_time(), runId=self.ctx.run_id
                     )
+                    if self.config.set_system_metadata_pipeline_name:
+                        workunit.metadata.systemMetadata.pipelineName = (
+                            self.ctx.pipeline_name
+                        )
                 if (
                     isinstance(workunit.metadata, MetadataChangeEvent)
                     and len(workunit.metadata.proposedSnapshot.aspects) == 0

diff --git a/metadata-models/src/main/pegasus/com/linkedin/mxe/SystemMetadata.pdl b/metadata-models/src/main/pegasus/com/linkedin/mxe/SystemMetadata.pdl
@@ -14,6 +14,11 @@ record SystemMetadata {
   */
   runId: optional string = "no-run-id-provided"
 
+  /**
+   * The ingestion pipeline id that produced the metadata. Populated in case of batch ingestion.
+   */
+  pipelineName: optional string
+
   /**
   * The model registry name that was used to process this event
   */