diff --git a/metadata-ingestion/docs/sources/nifi/nifi_pre.md b/metadata-ingestion/docs/sources/nifi/nifi_pre.md new file mode 100644 index 0000000000..5774c1bbba --- /dev/null +++ b/metadata-ingestion/docs/sources/nifi/nifi_pre.md @@ -0,0 +1,17 @@ +### Concept Mapping + +| Source Concept | DataHub Concept | Notes | +| --------------------------------- | --------------------------------------------------------- | ----------------------- | +| `"Nifi"` | [Data Platform](../../metamodel/entities/dataPlatform.md) | | +| Nifi flow | [Data Flow](../../metamodel/entities/dataFlow.md) | | +| Nifi Ingress / Egress Processor | [Data Job](../../metamodel/entities/dataJob.md) | | +| Nifi Remote Port | [Data Job](../../metamodel/entities/dataJob.md) | | +| Nifi Port with remote connections | [Dataset](../../metamodel/entities/dataset.md) | | +| Nifi Process Group | [Container](../../metamodel/entities/container.md) | Subtype `Process Group` | + +### Caveats +- This plugin extracts the lineage information between external datasets and ingress/egress processors by analyzing provenance events. Please check your Nifi configuration to confirm max rentention period of provenance events and make sure that ingestion runs frequent enough to read provenance events before they are disappear. + +- Limited ingress/egress processors are supported + - S3: `ListS3`, `FetchS3Object`, `PutS3Object` + - SFTP: `ListSFTP`, `FetchSFTP`, `GetSFTP`, `PutSFTP` \ No newline at end of file diff --git a/metadata-ingestion/src/datahub/ingestion/source/common/subtypes.py b/metadata-ingestion/src/datahub/ingestion/source/common/subtypes.py index 7bf038b0e6..b81a9b8ae6 100644 --- a/metadata-ingestion/src/datahub/ingestion/source/common/subtypes.py +++ b/metadata-ingestion/src/datahub/ingestion/source/common/subtypes.py @@ -45,6 +45,10 @@ class BIContainerSubTypes(str, Enum): QLIK_APP = "Qlik App" +class JobContainerSubTypes(str, Enum): + NIFI_PROCESS_GROUP = "Process Group" + + class BIAssetSubTypes(str, Enum): # Generic SubTypes REPORT = "Report" diff --git a/metadata-ingestion/src/datahub/ingestion/source/nifi.py b/metadata-ingestion/src/datahub/ingestion/source/nifi.py index d7b9c68542..d650a9b1d1 100644 --- a/metadata-ingestion/src/datahub/ingestion/source/nifi.py +++ b/metadata-ingestion/src/datahub/ingestion/source/nifi.py @@ -23,6 +23,7 @@ import datahub.emitter.mce_builder as builder from datahub.configuration.common import AllowDenyPattern from datahub.configuration.source_common import EnvConfigMixin from datahub.emitter.mcp import MetadataChangeProposalWrapper +from datahub.emitter.mcp_builder import ContainerKey, gen_containers from datahub.ingestion.api.common import PipelineContext from datahub.ingestion.api.decorators import ( SupportStatus, @@ -33,13 +34,17 @@ from datahub.ingestion.api.decorators import ( ) from datahub.ingestion.api.source import Source, SourceCapability, SourceReport from datahub.ingestion.api.workunit import MetadataWorkUnit +from datahub.ingestion.source.common.subtypes import JobContainerSubTypes from datahub.metadata.schema_classes import ( + BrowsePathEntryClass, + BrowsePathsV2Class, DataFlowInfoClass, DataJobInfoClass, DataJobInputOutputClass, DataPlatformInstanceClass, DatasetPropertiesClass, ) +from datahub.specific.datajob import DataJobPatchBuilder logger = logging.getLogger(__name__) NIFI = "nifi" @@ -70,6 +75,10 @@ class NifiAuthType(Enum): BASIC_AUTH = "BASIC_AUTH" +class ProcessGroupKey(ContainerKey): + process_group_id: str + + class NifiSourceConfig(EnvConfigMixin): site_url: str = Field( description="URL for Nifi, ending with /nifi/. e.g. https://mynifi.domain/nifi/" @@ -123,7 +132,21 @@ class NifiSourceConfig(EnvConfigMixin): # root CA trusted by client system, e.g. self-signed certificates ca_file: Optional[Union[bool, str]] = Field( default=None, - description="Path to PEM file containing certs for the root CA(s) for the NiFi", + description="Path to PEM file containing certs for the root CA(s) for the NiFi." + "Set to False to disable SSL verification.", + ) + + # As of now, container entities retrieval does not respect browsePathsV2 similar to container aspect. + # Consider enabling this when entities with browsePathsV2 pointing to container also get listed in container entities. + emit_process_group_as_container: bool = Field( + default=False, + description="Whether to emit Nifi process groups as container entities.", + ) + + incremental_lineage: bool = Field( + default=True, + description="When enabled, emits incremental/patch lineage for Nifi processors." + " When disabled, re-states lineage on each run.", ) @root_validator(skip_on_failure=True) @@ -364,21 +387,6 @@ class NifiSourceReport(SourceReport): @support_status(SupportStatus.CERTIFIED) @capability(SourceCapability.LINEAGE_COARSE, "Supported. See docs for limitations") class NifiSource(Source): - """ - This plugin extracts the following: - - - NiFi flow as `DataFlow` entity - - Ingress, egress processors, remote input and output ports as `DataJob` entity - - Input and output ports receiving remote connections as `Dataset` entity - - Lineage information between external datasets and ingress/egress processors by analyzing provenance events - - Current limitations: - - - Limited ingress/egress processors are supported - - S3: `ListS3`, `FetchS3Object`, `PutS3Object` - - SFTP: `ListSFTP`, `FetchSFTP`, `GetSFTP`, `PutSFTP` - - """ config: NifiSourceConfig report: NifiSourceReport @@ -392,6 +400,10 @@ class NifiSource(Source): if self.config.ca_file is not None: self.session.verify = self.config.ca_file + # To keep track of process groups (containers) which have already been ingested + # Required, as we do not ingest all process groups but only those that have known ingress/egress processors + self.processed_pgs: List[str] = [] + @cached_property def rest_api_base_url(self): return self.config.site_url[: -len("nifi/")] + "nifi-api/" @@ -794,7 +806,7 @@ class NifiSource(Source): def construct_workunits(self) -> Iterable[MetadataWorkUnit]: # noqa: C901 rootpg = self.nifi_flow.root_process_group flow_name = rootpg.name # self.config.site_name - flow_urn = builder.make_data_flow_urn(NIFI, rootpg.id, self.config.env) + flow_urn = self.make_flow_urn() flow_properties = {} if self.nifi_flow.clustered is not None: flow_properties["clustered"] = str(self.nifi_flow.clustered) @@ -927,9 +939,16 @@ class NifiSource(Source): ) break + if self.config.emit_process_group_as_container: + # We emit process groups only for all nifi components qualifying as datajobs + yield from self.construct_process_group_workunits( + component.parent_group_id + ) + yield from self.construct_job_workunits( job_urn, job_name, + component.parent_group_id, external_url=self.make_external_url( component.parent_group_id, component.id, component.parent_rpg_id ), @@ -951,6 +970,11 @@ class NifiSource(Source): external_url=self.make_external_url(port.parent_group_id, port.id), ) + def make_flow_urn(self) -> str: + return builder.make_data_flow_urn( + NIFI, self.nifi_flow.root_process_group.id, self.config.env + ) + def process_provenance_events(self): startDate = datetime.now(timezone.utc) - timedelta( days=self.config.provenance_days @@ -1083,6 +1107,7 @@ class NifiSource(Source): self, job_urn: str, job_name: str, + parent_group_id: str, external_url: str, job_type: str, description: Optional[str], @@ -1107,17 +1132,114 @@ class NifiSource(Source): ), ).as_workunit() + # If dataJob had container aspect, we would ideally only emit it + # and browse path v2 would automatically be generated. + yield self.gen_browse_path_v2_workunit(job_urn, parent_group_id) + inlets.sort() outlets.sort() inputJobs.sort() - yield MetadataChangeProposalWrapper( - entityUrn=job_urn, - aspect=DataJobInputOutputClass( - inputDatasets=inlets, outputDatasets=outlets, inputDatajobs=inputJobs + if self.config.incremental_lineage: + patch_builder: DataJobPatchBuilder = DataJobPatchBuilder(job_urn) + for inlet in inlets: + patch_builder.add_input_dataset(inlet) + for outlet in outlets: + patch_builder.add_output_dataset(outlet) + for inJob in inputJobs: + patch_builder.add_input_datajob(inJob) + for patch_mcp in patch_builder.build(): + yield MetadataWorkUnit( + id=f"{job_urn}-{patch_mcp.aspectName}", mcp_raw=patch_mcp + ) + else: + yield MetadataChangeProposalWrapper( + entityUrn=job_urn, + aspect=DataJobInputOutputClass( + inputDatasets=inlets, + outputDatasets=outlets, + inputDatajobs=inputJobs, + ), + ).as_workunit() + + def gen_browse_path_v2_workunit( + self, entity_urn: str, process_group_id: str + ) -> MetadataWorkUnit: + flow_urn = self.make_flow_urn() + return MetadataChangeProposalWrapper( + entityUrn=entity_urn, + aspect=BrowsePathsV2Class( + path=[ + BrowsePathEntryClass(id=flow_urn, urn=flow_urn), + *self._get_browse_path_v2_entries(process_group_id), + ] ), ).as_workunit() + def _get_browse_path_v2_entries( + self, process_group_id: str + ) -> List[BrowsePathEntryClass]: + """Browse path entries till current process group""" + if self._is_root_process_group(process_group_id): + return [] + + current_process_group = self.nifi_flow.processGroups[process_group_id] + assert ( + current_process_group.parent_group_id + ) # always present for non-root process group + parent_browse_path = self._get_browse_path_v2_entries( + current_process_group.parent_group_id + ) + + if self.config.emit_process_group_as_container: + container_urn = self.gen_process_group_key(process_group_id).as_urn() + current_browse_entry = BrowsePathEntryClass( + id=container_urn, urn=container_urn + ) + else: + current_browse_entry = BrowsePathEntryClass(id=current_process_group.name) + return parent_browse_path + [current_browse_entry] + + def _is_root_process_group(self, process_group_id: str) -> bool: + return self.nifi_flow.root_process_group.id == process_group_id + + def construct_process_group_workunits( + self, process_group_id: str + ) -> Iterable[MetadataWorkUnit]: + if ( + self._is_root_process_group(process_group_id) + or process_group_id in self.processed_pgs + ): + return + self.processed_pgs.append(process_group_id) + + pg = self.nifi_flow.processGroups[process_group_id] + container_key = self.gen_process_group_key(process_group_id) + yield from gen_containers( + container_key=container_key, + name=pg.name, + sub_types=[JobContainerSubTypes.NIFI_PROCESS_GROUP], + parent_container_key=( + self.gen_process_group_key(pg.parent_group_id) + if pg.parent_group_id + and not self._is_root_process_group(pg.parent_group_id) + else None + ), + ) + + if pg.parent_group_id: # always true for non-root process group + yield from self.construct_process_group_workunits(pg.parent_group_id) + + if self._is_root_process_group(pg.parent_group_id): + yield self.gen_browse_path_v2_workunit( + container_key.as_urn(), pg.parent_group_id + ) + + def gen_process_group_key(self, process_group_id: str) -> ProcessGroupKey: + return ProcessGroupKey( + process_group_id=process_group_id, platform=NIFI, env=self.config.env + ) + def construct_dataset_workunits( self, dataset_platform: str, diff --git a/metadata-ingestion/tests/integration/nifi/nifi_mces_golden_cluster.json b/metadata-ingestion/tests/integration/nifi/nifi_mces_golden_cluster.json index e0ab6bbec7..7a50344a1d 100644 --- a/metadata-ingestion/tests/integration/nifi/nifi_mces_golden_cluster.json +++ b/metadata-ingestion/tests/integration/nifi/nifi_mces_golden_cluster.json @@ -16,7 +16,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -44,7 +45,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -63,7 +65,32 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" + } +}, +{ + "entityType": "dataJob", + "entityUrn": "urn:li:dataJob:(urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD),fed5914b-937b-37dd-89c0-b34ffbae9cf4)", + "changeType": "UPSERT", + "aspectName": "browsePathsV2", + "aspect": { + "json": { + "path": [ + { + "id": "urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD)", + "urn": "urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD)" + }, + { + "id": "Cluster_Site_S3_to_S3" + } + ] + } + }, + "systemMetadata": { + "lastObserved": 1638532800000, + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -78,7 +105,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -96,7 +124,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -124,7 +153,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -145,7 +175,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -160,7 +191,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -178,7 +210,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -206,7 +239,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -225,7 +259,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -246,7 +281,32 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" + } +}, +{ + "entityType": "dataJob", + "entityUrn": "urn:li:dataJob:(urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD),c5f6fc66-ffbb-3f60-9564-f2466ae32493)", + "changeType": "UPSERT", + "aspectName": "browsePathsV2", + "aspect": { + "json": { + "path": [ + { + "id": "urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD)", + "urn": "urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD)" + }, + { + "id": "Cluster_Site_S3_to_S3" + } + ] + } + }, + "systemMetadata": { + "lastObserved": 1638532800000, + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -267,7 +327,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -295,7 +356,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -314,7 +376,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -329,7 +392,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -347,7 +411,32 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" + } +}, +{ + "entityType": "dataJob", + "entityUrn": "urn:li:dataJob:(urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD),8a218b6e-e6a0-36b6-bc4b-79d202a80167)", + "changeType": "UPSERT", + "aspectName": "browsePathsV2", + "aspect": { + "json": { + "path": [ + { + "id": "urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD)", + "urn": "urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD)" + }, + { + "id": "Cluster_Site_S3_to_S3" + } + ] + } + }, + "systemMetadata": { + "lastObserved": 1638532800000, + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -362,7 +451,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -380,7 +470,56 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" + } +}, +{ + "entityType": "dataJob", + "entityUrn": "urn:li:dataJob:(urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD),71bc17ed-a3bc-339a-a100-ebad434717d4)", + "changeType": "UPSERT", + "aspectName": "browsePathsV2", + "aspect": { + "json": { + "path": [ + { + "id": "urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD)", + "urn": "urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD)" + }, + { + "id": "Cluster_Site_S3_to_S3" + } + ] + } + }, + "systemMetadata": { + "lastObserved": 1638532800000, + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" + } +}, +{ + "entityType": "dataJob", + "entityUrn": "urn:li:dataJob:(urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD),c8c73d4c-ebdd-1bee-9b46-629672cd11a0)", + "changeType": "UPSERT", + "aspectName": "browsePathsV2", + "aspect": { + "json": { + "path": [ + { + "id": "urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD)", + "urn": "urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD)" + }, + { + "id": "Cluster_Site_SFTP_to_S3" + } + ] + } + }, + "systemMetadata": { + "lastObserved": 1638532800000, + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -408,7 +547,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -428,7 +568,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -443,7 +584,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -461,7 +603,32 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" + } +}, +{ + "entityType": "dataJob", + "entityUrn": "urn:li:dataJob:(urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD),8eb5263d-017d-1000-ffff-ffff911b23aa)", + "changeType": "UPSERT", + "aspectName": "browsePathsV2", + "aspect": { + "json": { + "path": [ + { + "id": "urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD)", + "urn": "urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD)" + }, + { + "id": "Cluster_Site_SFTP_to_S3" + } + ] + } + }, + "systemMetadata": { + "lastObserved": 1638532800000, + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -476,7 +643,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -494,7 +662,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -522,7 +691,32 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" + } +}, +{ + "entityType": "dataJob", + "entityUrn": "urn:li:dataJob:(urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD),8eb55aeb-017d-1000-ffff-fffff475768d)", + "changeType": "UPSERT", + "aspectName": "browsePathsV2", + "aspect": { + "json": { + "path": [ + { + "id": "urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD)", + "urn": "urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD)" + }, + { + "id": "Cluster_Site_SFTP_to_S3" + } + ] + } + }, + "systemMetadata": { + "lastObserved": 1638532800000, + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -546,7 +740,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -567,7 +762,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -586,7 +782,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -601,7 +798,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -618,7 +816,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -633,7 +832,32 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" + } +}, +{ + "entityType": "dataJob", + "entityUrn": "urn:li:dataJob:(urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD),3ec2acd6-a0d4-3198-9066-a59fb757bc05)", + "changeType": "UPSERT", + "aspectName": "browsePathsV2", + "aspect": { + "json": { + "path": [ + { + "id": "urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD)", + "urn": "urn:li:dataFlow:(nifi,80820b2f-017d-1000-85cf-05f56cde9185,PROD)" + }, + { + "id": "Cluster_Site_SFTP_to_S3" + } + ] + } + }, + "systemMetadata": { + "lastObserved": 1638532800000, + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -650,7 +874,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -665,7 +890,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -680,7 +906,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -695,7 +922,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -710,7 +938,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -725,7 +954,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -740,7 +970,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -755,7 +986,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -770,7 +1002,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -785,7 +1018,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -800,7 +1034,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -815,7 +1050,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -830,7 +1066,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -845,7 +1082,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } }, { @@ -860,7 +1098,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-cluster" + "runId": "nifi-test-cluster", + "lastRunId": "no-run-id-provided" } } ] \ No newline at end of file diff --git a/metadata-ingestion/tests/integration/nifi/nifi_mces_golden_standalone.json b/metadata-ingestion/tests/integration/nifi/nifi_mces_golden_standalone.json index 48a22bebb3..f820efc739 100644 --- a/metadata-ingestion/tests/integration/nifi/nifi_mces_golden_standalone.json +++ b/metadata-ingestion/tests/integration/nifi/nifi_mces_golden_standalone.json @@ -16,7 +16,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-standalone" + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" } }, { @@ -44,26 +45,62 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-standalone" + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" + } +}, +{ + "entityType": "dataJob", + "entityUrn": "urn:li:dataJob:(urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD),aed63edf-e660-3f29-b56b-192cf6286889)", + "changeType": "PATCH", + "aspectName": "dataJobInputOutput", + "aspect": { + "json": [ + { + "op": "add", + "path": "/inputDatajobEdges/urn:li:dataJob:(urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD),91d59f03-1c2b-3f3f-48bc-f89296a328bd)", + "value": { + "destinationUrn": "urn:li:dataJob:(urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD),91d59f03-1c2b-3f3f-48bc-f89296a328bd)", + "created": { + "time": 1638532800000, + "actor": "urn:li:corpuser:datahub" + }, + "lastModified": { + "time": 1638532800000, + "actor": "urn:li:corpuser:datahub" + } + } + } + ] + }, + "systemMetadata": { + "lastObserved": 1638532800000, + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" } }, { "entityType": "dataJob", "entityUrn": "urn:li:dataJob:(urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD),aed63edf-e660-3f29-b56b-192cf6286889)", "changeType": "UPSERT", - "aspectName": "dataJobInputOutput", + "aspectName": "browsePathsV2", "aspect": { "json": { - "inputDatasets": [], - "outputDatasets": [], - "inputDatajobs": [ - "urn:li:dataJob:(urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD),91d59f03-1c2b-3f3f-48bc-f89296a328bd)" + "path": [ + { + "id": "urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD)", + "urn": "urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD)" + }, + { + "id": "Single_Site_S3_to_S3" + } ] } }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-standalone" + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" } }, { @@ -78,7 +115,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-standalone" + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" } }, { @@ -96,7 +134,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-standalone" + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" } }, { @@ -124,28 +163,53 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-standalone" + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" } }, { "entityType": "dataJob", "entityUrn": "urn:li:dataJob:(urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD),91d59f03-1c2b-3f3f-48bc-f89296a328bd)", - "changeType": "UPSERT", + "changeType": "PATCH", "aspectName": "dataJobInputOutput", "aspect": { - "json": { - "inputDatasets": [ - "urn:li:dataset:(urn:li:dataPlatform:s3,enriched-topical-chat,PROD)" - ], - "outputDatasets": [], - "inputDatajobs": [ - "urn:li:dataJob:(urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD),cb7693ed-f93b-3340-3776-fe80e6283ddc)" - ] - } + "json": [ + { + "op": "add", + "path": "/inputDatasetEdges/urn:li:dataset:(urn:li:dataPlatform:s3,enriched-topical-chat,PROD)", + "value": { + "destinationUrn": "urn:li:dataset:(urn:li:dataPlatform:s3,enriched-topical-chat,PROD)", + "created": { + "time": 1638532800000, + "actor": "urn:li:corpuser:datahub" + }, + "lastModified": { + "time": 1638532800000, + "actor": "urn:li:corpuser:datahub" + } + } + }, + { + "op": "add", + "path": "/inputDatajobEdges/urn:li:dataJob:(urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD),cb7693ed-f93b-3340-3776-fe80e6283ddc)", + "value": { + "destinationUrn": "urn:li:dataJob:(urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD),cb7693ed-f93b-3340-3776-fe80e6283ddc)", + "created": { + "time": 1638532800000, + "actor": "urn:li:corpuser:datahub" + }, + "lastModified": { + "time": 1638532800000, + "actor": "urn:li:corpuser:datahub" + } + } + } + ] }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-standalone" + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" } }, { @@ -160,7 +224,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-standalone" + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" } }, { @@ -178,7 +243,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-standalone" + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" } }, { @@ -206,26 +272,38 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-standalone" + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" } }, { "entityType": "dataJob", "entityUrn": "urn:li:dataJob:(urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD),cb7693ed-f93b-3340-3776-fe80e6283ddc)", - "changeType": "UPSERT", + "changeType": "PATCH", "aspectName": "dataJobInputOutput", "aspect": { - "json": { - "inputDatasets": [ - "urn:li:dataset:(urn:li:dataPlatform:s3,enriched-topical-chat,PROD)" - ], - "outputDatasets": [], - "inputDatajobs": [] - } + "json": [ + { + "op": "add", + "path": "/inputDatasetEdges/urn:li:dataset:(urn:li:dataPlatform:s3,enriched-topical-chat,PROD)", + "value": { + "destinationUrn": "urn:li:dataset:(urn:li:dataPlatform:s3,enriched-topical-chat,PROD)", + "created": { + "time": 1638532800000, + "actor": "urn:li:corpuser:datahub" + }, + "lastModified": { + "time": 1638532800000, + "actor": "urn:li:corpuser:datahub" + } + } + } + ] }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-standalone" + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" } }, { @@ -240,7 +318,32 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-standalone" + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" + } +}, +{ + "entityType": "dataJob", + "entityUrn": "urn:li:dataJob:(urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD),91d59f03-1c2b-3f3f-48bc-f89296a328bd)", + "changeType": "UPSERT", + "aspectName": "browsePathsV2", + "aspect": { + "json": { + "path": [ + { + "id": "urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD)", + "urn": "urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD)" + }, + { + "id": "Single_Site_S3_to_S3" + } + ] + } + }, + "systemMetadata": { + "lastObserved": 1638532800000, + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" } }, { @@ -255,7 +358,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-standalone" + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" } }, { @@ -270,7 +374,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-standalone" + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" } }, { @@ -285,7 +390,8 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-standalone" + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" } }, { @@ -300,7 +406,32 @@ }, "systemMetadata": { "lastObserved": 1638532800000, - "runId": "nifi-test-standalone" + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" + } +}, +{ + "entityType": "dataJob", + "entityUrn": "urn:li:dataJob:(urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD),cb7693ed-f93b-3340-3776-fe80e6283ddc)", + "changeType": "UPSERT", + "aspectName": "browsePathsV2", + "aspect": { + "json": { + "path": [ + { + "id": "urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD)", + "urn": "urn:li:dataFlow:(nifi,803ebb92-017d-1000-2961-4bdaa27a3ba0,PROD)" + }, + { + "id": "Single_Site_S3_to_S3" + } + ] + } + }, + "systemMetadata": { + "lastObserved": 1638532800000, + "runId": "nifi-test-standalone", + "lastRunId": "no-run-id-provided" } } ] \ No newline at end of file diff --git a/metadata-ingestion/tests/integration/nifi/test_nifi.py b/metadata-ingestion/tests/integration/nifi/test_nifi.py index bf17ee7472..b992de0588 100644 --- a/metadata-ingestion/tests/integration/nifi/test_nifi.py +++ b/metadata-ingestion/tests/integration/nifi/test_nifi.py @@ -143,6 +143,7 @@ def test_nifi_ingest_cluster(loaded_nifi, pytestconfig, tmp_path, test_resources "http://nifi02:9081/nifi/": "default", "http://nifi03:9082/nifi/": "default", }, + "incremental_lineage": False, }, }, "sink": { diff --git a/metadata-ingestion/tests/unit/test_nifi_source.py b/metadata-ingestion/tests/unit/test_nifi_source.py index 5d7949c04b..26c803927f 100644 --- a/metadata-ingestion/tests/unit/test_nifi_source.py +++ b/metadata-ingestion/tests/unit/test_nifi_source.py @@ -17,9 +17,7 @@ from datahub.ingestion.source.nifi import ( @typing.no_type_check def test_nifi_s3_provenance_event(): - config_dict = { - "site_url": "http://localhost:8080", - } + config_dict = {"site_url": "http://localhost:8080", "incremental_lineage": False} nifi_config = NifiSourceConfig.parse_obj(config_dict) ctx = PipelineContext(run_id="test") @@ -79,15 +77,16 @@ def test_nifi_s3_provenance_event(): # one aspect for dataflow and two aspects for datajob # and two aspects for dataset - assert len(workunits) == 5 + assert len(workunits) == 6 assert workunits[0].metadata.entityType == "dataFlow" assert workunits[1].metadata.entityType == "dataset" assert workunits[2].metadata.entityType == "dataset" assert workunits[3].metadata.entityType == "dataJob" assert workunits[4].metadata.entityType == "dataJob" + assert workunits[5].metadata.entityType == "dataJob" - ioAspect = workunits[4].metadata.aspect + ioAspect = workunits[5].metadata.aspect assert ioAspect.outputDatasets == [ "urn:li:dataset:(urn:li:dataPlatform:s3,foo-nifi/tropical_data,PROD)" ] @@ -149,15 +148,16 @@ def test_nifi_s3_provenance_event(): # one aspect for dataflow and two aspects for datajob # and two aspects for dataset - assert len(workunits) == 5 + assert len(workunits) == 6 assert workunits[0].metadata.entityType == "dataFlow" assert workunits[1].metadata.entityType == "dataset" assert workunits[2].metadata.entityType == "dataset" assert workunits[3].metadata.entityType == "dataJob" assert workunits[4].metadata.entityType == "dataJob" + assert workunits[5].metadata.entityType == "dataJob" - ioAspect = workunits[4].metadata.aspect + ioAspect = workunits[5].metadata.aspect assert ioAspect.outputDatasets == [] assert ioAspect.inputDatasets == [ "urn:li:dataset:(urn:li:dataPlatform:s3,enriched-topical-chat,PROD)"