Skip to content

Commit f9e8165

Browse files
dolfinusdolfinus
committed
[DOP-38389] Spark job.name=unknown should be detached from others
1 parent 6e1934a commit f9e8165

7 files changed

Lines changed: 270 additions & 6 deletions

File tree

‎data_rentgen/consumer/extractors/impl/spark.py‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,12 @@ def is_operation(self, event: OpenLineageRunEvent) -> bool:
3838
def extract_run(self, event: OpenLineageRunEvent) -> RunDTO:
3939
run = super().extract_run(event)
4040
self._enrich_run_identifiers(run, event)
41+
if run.job.name == "unknown":
42+
# Workaround for https://github.com/OpenLineage/OpenLineage/issues/3846
43+
# Avoid updating the same temporary job multiple times
44+
run.job.parent_job = None
45+
run.job.tag_values.clear()
46+
run.job_dependencies.clear()
4147
return run
4248

4349
def _enrich_run_identifiers(self, run: RunDTO, event: OpenLineageRunEvent):
@@ -60,7 +66,6 @@ def extract_operation(self, event: OpenLineageRunEvent) -> OperationDTO:
6066
# For Spark, SQL_JOB --parent-> SPARK_APPLICATION = operation -> run,
6167
# and parent is always here.
6268
run = self.extract_parent_run(event.run.facets.parent) # type: ignore[arg-type]
63-
# Workaround for https://github.com/OpenLineage/OpenLineage/issues/3846
6469
self._enrich_run_identifiers(run, event)
6570
self._enrich_run_tags(run, event)
6671
operation = super()._extract_operation(event, run)

‎data_rentgen/db/migrations/versions/2026-07-06_96fd9a096682_detach_spark_unknown_jobs_from_.py‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
# SPDX-FileCopyrightText: 2024-present MTS PJSC
2+
# SPDX-License-Identifier: Apache-2.0
3+
"""Detach Spark unknown jobs from everything
4+
5+
It should have no parent job, no dependencies and no tags.
6+
7+
Revision ID: 96fd9a096682
8+
Revises: 4a02d2d5c8b1
9+
Create Date: 2026-07-06 13:01:46.138108
10+
11+
"""
12+
13+
import sqlalchemy as sa
14+
from alembic import op
15+
16+
# revision identifiers, used by Alembic.
17+
revision = "96fd9a096682"
18+
down_revision = "4a02d2d5c8b1"
19+
branch_labels = None
20+
depends_on = None
21+
22+
SPARK_UNKNOWN_JOBS_QUERY = """
23+
SELECT job.*
24+
FROM job
25+
JOIN job_type ON job.type_id = job_type.id
26+
WHERE job_type.type = 'SPARK_APPLICATION'
27+
AND lower(job.name) = 'unknown'
28+
"""
29+
30+
31+
def upgrade() -> None:
32+
op.execute(
33+
sa.text(
34+
f"""
35+
WITH data AS ({SPARK_UNKNOWN_JOBS_QUERY})
36+
UPDATE job
37+
SET parent_job_id=NULL
38+
WHERE id IN (SELECT id FROM data);
39+
""",
40+
),
41+
)
42+
op.execute(
43+
sa.text(
44+
f"""
45+
WITH data AS ({SPARK_UNKNOWN_JOBS_QUERY})
46+
DELETE FROM job_dependency
47+
WHERE from_job_id IN (SELECT id FROM data) OR to_job_id IN (SELECT id FROM data);
48+
""",
49+
),
50+
)
51+
op.execute(
52+
sa.text(
53+
f"""
54+
WITH data AS ({SPARK_UNKNOWN_JOBS_QUERY})
55+
DELETE FROM job_tag_value
56+
WHERE job_id IN (SELECT id FROM data);
57+
""",
58+
),
59+
)
60+
61+
62+
def downgrade() -> None:
63+
# this migration is irreversible
64+
pass

‎docs/changelog/next_release/480.improvement.rst‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
If Data.Rentgen receives RunEvent with Spark job with ``name=unknown``, it should have no parent job, no tags and no dependencies.
2+
Otherwise multiple workers will update the same row each time, although it is temporary.
3+
4+
Workaround for ``OpenLineage #3846<https://github.com/OpenLineage/OpenLineage/issues/3846>`_.

‎tests/test_consumer/test_extractors/fixtures/spark_raw.py‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -179,7 +179,7 @@ def spark_app_run_event_start() -> OpenLineageRunEvent:
179179
facets=OpenLineageRunFacets(
180180
spark_applicationDetails=OpenLineageSparkApplicationDetailsRunFacet(
181181
master="local[*]",
182-
appName="spark_session",
182+
appName="mysession1",
183183
applicationId="local-1719136537510",
184184
deployMode=OpenLineageSparkDeployMode.CLIENT,
185185
driverHost="127.0.0.1",
@@ -400,7 +400,7 @@ def spark_operation_run_event_start() -> OpenLineageRunEvent:
400400
),
401401
spark_applicationDetails=OpenLineageSparkApplicationDetailsRunFacet(
402402
master="local[*]",
403-
appName="spark_session",
403+
appName="mysession1",
404404
applicationId="local-1719136537510",
405405
deployMode=OpenLineageSparkDeployMode.CLIENT,
406406
driverHost="127.0.0.1",
@@ -445,7 +445,7 @@ def spark_operation_run_event_running() -> OpenLineageRunEvent:
445445
),
446446
spark_applicationDetails=OpenLineageSparkApplicationDetailsRunFacet(
447447
master="local[*]",
448-
appName="spark_session",
448+
appName="mysession1",
449449
applicationId="local-1719136537510",
450450
deployMode=OpenLineageSparkDeployMode.CLIENT,
451451
driverHost="127.0.0.1",

‎tests/test_consumer/test_extractors/test_extractors_batch_spark.py‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -167,8 +167,6 @@ def test_extractors_extract_batch_spark_openlineage_emitted_unknown_name_no_job_
167167

168168
extracted = BatchExtractor().add_events(events)
169169

170-
extracted_spark_unknown_job.parent_job = extracted_airflow_task1_job_as_parent
171-
172170
# the result is the same as above
173171
assert extracted.locations() == [
174172
extracted_airflow_location,

‎tests/test_consumer/test_extractors/test_extractors_run_spark.py‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -370,3 +370,75 @@ def test_extractors_extract_run_spark_parent_job():
370370
persistent_log_url=None,
371371
running_log_url="http://localhost:4040",
372372
)
373+
374+
375+
def test_extractors_extract_run_spark_parent_job_for_unknown():
376+
now = datetime(2024, 7, 5, 9, 4, 13, 979349, tzinfo=timezone.utc)
377+
run_id = UUID("01908223-0e9b-7c52-9856-6cecfc842610")
378+
parent_run_id = UUID("01908224-8410-79a2-8de6-a769ad6944c9")
379+
run = OpenLineageRunEvent(
380+
eventType=OpenLineageRunEventType.RUNNING,
381+
eventTime=now,
382+
job=OpenLineageJob(
383+
namespace="host://some.host.com",
384+
name="unknown",
385+
facets=OpenLineageJobFacets(
386+
jobType=OpenLineageJobTypeJobFacet(
387+
processingType=OpenLineageJobProcessingType.NONE,
388+
integration="SPARK",
389+
jobType="APPLICATION",
390+
),
391+
),
392+
),
393+
run=OpenLineageRun(
394+
runId=run_id,
395+
facets=OpenLineageRunFacets(
396+
parent=OpenLineageParentRunFacet(
397+
job=OpenLineageParentJob(
398+
namespace="anything",
399+
name="parentjob",
400+
),
401+
run=OpenLineageParentRun(
402+
runId=parent_run_id,
403+
),
404+
),
405+
processing_engine=OpenLineageProcessingEngineRunFacet(
406+
version=Version("3.4.3"),
407+
name="spark",
408+
openlineageAdapterVersion=Version("1.19.0"),
409+
),
410+
),
411+
),
412+
)
413+
414+
assert SparkExtractor().extract_run(run) == RunDTO(
415+
id=run_id,
416+
job=JobDTO(
417+
name="unknown",
418+
parent_job=None,
419+
location=LocationDTO(type="host", name="some.host.com", addresses={"host://some.host.com"}),
420+
type=JobTypeDTO(type="SPARK_APPLICATION"),
421+
tag_values=set(),
422+
),
423+
parent_run=RunDTO(
424+
id=parent_run_id,
425+
job=JobDTO(
426+
name="parentjob",
427+
type=None,
428+
location=LocationDTO(
429+
type="unknown",
430+
name="anything",
431+
addresses={"unknown://anything"},
432+
),
433+
),
434+
status=RunStatusDTO.UNKNOWN,
435+
),
436+
status=RunStatusDTO.STARTED,
437+
started_at=None,
438+
start_reason=None,
439+
user=None,
440+
external_id=None,
441+
attempt=None,
442+
persistent_log_url=None,
443+
running_log_url=None,
444+
)

‎tests/test_database/test_migration_detach_spark_unknown_jobs.py‎

Lines changed: 121 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,121 @@
1+
# SPDX-FileCopyrightText: 2024-present MTS PJSC
2+
# SPDX-License-Identifier: Apache-2.0
3+
from __future__ import annotations
4+
5+
from typing import TYPE_CHECKING
6+
7+
import pytest
8+
from sqlalchemy import create_engine, text
9+
from sqlalchemy.pool import NullPool
10+
11+
from data_rentgen.db.models import Base
12+
from tests.test_database.fixtures.alembic import do_run_migrations
13+
14+
if TYPE_CHECKING:
15+
from alembic.config import Config as AlembicConfig
16+
17+
pytestmark = [pytest.mark.db]
18+
19+
PREV_REVISION = "4a02d2d5c8b1"
20+
THIS_REVISION = "96fd9a096682"
21+
22+
23+
def test_migration_detach_spark_unknown_jobs(empty_db_url: str, alembic_config: AlembicConfig):
24+
do_run_migrations(alembic_config, Base.metadata, PREV_REVISION)
25+
26+
engine = create_engine(empty_db_url, poolclass=NullPool)
27+
try:
28+
with engine.begin() as conn:
29+
conn.execute(
30+
text(
31+
"""
32+
INSERT INTO location (id, type, name) VALUES
33+
(1, 'local', 'somehost');
34+
""",
35+
),
36+
)
37+
38+
conn.execute(
39+
text(
40+
"""
41+
INSERT INTO job (id, location_id, name, type_id, parent_job_id) VALUES
42+
(1, 1, 'airflow_task1', 3, null),
43+
(2, 1, 'airflow_task2', 3, null),
44+
(3, 1, 'unknown', 1, 1),
45+
(4, 1, 'some_session', 1, 2);
46+
""",
47+
),
48+
)
49+
50+
conn.execute(
51+
text(
52+
"""
53+
INSERT INTO tag (id, name) VALUES
54+
(1, 'some_tag'),
55+
(2, 'another_tag');
56+
""",
57+
),
58+
)
59+
60+
conn.execute(
61+
text(
62+
"""
63+
INSERT INTO tag_value (id, tag_id, value) VALUES
64+
(1, 1, 'some_value'),
65+
(2, 2, 'another_value');
66+
""",
67+
),
68+
)
69+
70+
conn.execute(
71+
text(
72+
"""
73+
INSERT INTO job_tag_value (job_id, tag_value_id) VALUES
74+
(1, 1),
75+
(2, 1),
76+
(3, 2),
77+
(4, 2);
78+
""",
79+
),
80+
)
81+
82+
conn.execute(
83+
text(
84+
"""
85+
INSERT INTO job_dependency (from_job_id, to_job_id) VALUES
86+
(1, 2),
87+
(3, 4);
88+
""",
89+
),
90+
)
91+
92+
do_run_migrations(alembic_config, Base.metadata, THIS_REVISION)
93+
94+
with engine.connect() as conn:
95+
assert conn.execute(
96+
text("SELECT id, parent_job_id FROM job ORDER BY id"),
97+
).fetchall() == [
98+
(1, None),
99+
(2, None),
100+
(3, None), # parent_id is null
101+
(4, 2),
102+
]
103+
104+
assert conn.execute(
105+
text("SELECT job_id, tag_value_id FROM job_tag_value ORDER BY job_id, tag_value_id"),
106+
).fetchall() == [
107+
(1, 1),
108+
(2, 1),
109+
# no item (3, 2)
110+
(4, 2),
111+
]
112+
113+
assert conn.execute(
114+
text("SELECT from_job_id, to_job_id FROM job_dependency ORDER BY from_job_id, to_job_id"),
115+
).fetchall() == [
116+
(1, 2),
117+
# no item (3, 4)
118+
]
119+
120+
finally:
121+
engine.dispose()

0 commit comments

Comments
 (0)