yujunjun/datahub
2
0
Fork 0
mirror of https://github.com/datahub-project/datahub.git synced 2025-10-31 02:37:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ingest/nifi): remove duplicate upstream jobs (#10849)

Browse Source
This commit is contained in:
Mayuri Nehate 2024-07-04 12:57:06 +05:30 committed by GitHub
parent 906bc98c82
commit 6e50bcc7ea
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 3 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
metadata-ingestion/src/datahub/ingestion/source/nifi.py
View File

@ -822,7 +822,7 @@ class NifiSource(Source):
outgoing = list( outgoing = list(
filter(lambda x: x[0] == component.id, self.nifi_flow.connections) filter(lambda x: x[0] == component.id, self.nifi_flow.connections)
) )
inputJobs = [] inputJobs = set()
jobProperties = None jobProperties = None
if component.nifi_type is NifiType.PROCESSOR: if component.nifi_type is NifiType.PROCESSOR:
@ -873,7 +873,7 @@ class NifiSource(Source):
dataset_urn, dataset_urn,
) )
else: else:
inputJobs.append( inputJobs.add(
builder.make_data_job_urn_with_flow(flow_urn, incoming_from) builder.make_data_job_urn_with_flow(flow_urn, incoming_from)
) )
@ -953,7 +953,7 @@ class NifiSource(Source):
job_properties=jobProperties, job_properties=jobProperties,
inlets=list(component.inlets.keys()), inlets=list(component.inlets.keys()),
outlets=list(component.outlets.keys()), outlets=list(component.outlets.keys()),
inputJobs=inputJobs, inputJobs=list(inputJobs),
status=component.status, status=component.status,
) )