From e3e9ab510563223471d948f3c75e06a5f6109786 Mon Sep 17 00:00:00 2001 From: tnazarew Date: Fri, 16 Oct 2026 16:05:28 +0200 Subject: [PATCH] Add hive_insert scenario and gate hive to OpenLineage 1.55.0 The hive scenario fails intermittently because GcsTransport names event files by eventTime in milliseconds and refuses to overwrite. The CTAS COMPLETE and its nested insert COMPLETE often share a millisecond, so one of them is dropped. The test is correct; the transport loses the event. hive_insert covers the same t1 -> t2 lineage with CREATE TABLE + INSERT INTO ... SELECT, which has no nested execution. Without CTAS, Spark 3.3.2 and 3.5.1 emit the same insert job, so one set of expected events covers all Spark versions. hive now runs from OpenLineage 1.55.0, by which time the transport should be fixed upstream. Co-Authored-By: Claude Opus 5.5 Signed-off-by: tnazarew --- .../spark_dataproc/scenarios/hive/config.json | 2 +- .../scenarios/hive_insert/config.json | 137 ++++++++++++++++++ .../hive_insert/events/columnLineage.json | 53 +++++++ .../hive_insert/events/dataSource.json | 31 ++++ .../hive_insert/events/gcp_dataproc.json | 17 +++ .../events/gcp_dataproc_spark.json | 17 +++ .../hive_insert/events/gcp_lineage.json | 15 ++ .../scenarios/hive_insert/events/jobType.json | 14 ++ .../scenarios/hive_insert/events/parent.json | 20 +++ .../hive_insert/events/processing_engine.json | 16 ++ .../hive_insert/events/run_event.json | 28 ++++ .../scenarios/hive_insert/events/schema.json | 47 ++++++ .../hive_insert/events/spark_properties.json | 17 +++ .../scenarios/hive_insert/events/sql.json | 12 ++ .../hive_insert/events/symlinks.json | 41 ++++++ .../scenarios/hive_insert/maintainers.json | 14 ++ .../scenarios/hive_insert/scenario.md | 34 +++++ .../scenarios/hive_insert/test/test.py | 19 +++ 18 files changed, 533 insertions(+), 1 deletion(-) create mode 100644 producer/spark_dataproc/scenarios/hive_insert/config.json create mode 100644 producer/spark_dataproc/scenarios/hive_insert/events/columnLineage.json create mode 100644 producer/spark_dataproc/scenarios/hive_insert/events/dataSource.json create mode 100644 producer/spark_dataproc/scenarios/hive_insert/events/gcp_dataproc.json create mode 100644 producer/spark_dataproc/scenarios/hive_insert/events/gcp_dataproc_spark.json create mode 100644 producer/spark_dataproc/scenarios/hive_insert/events/gcp_lineage.json create mode 100644 producer/spark_dataproc/scenarios/hive_insert/events/jobType.json create mode 100644 producer/spark_dataproc/scenarios/hive_insert/events/parent.json create mode 100644 producer/spark_dataproc/scenarios/hive_insert/events/processing_engine.json create mode 100644 producer/spark_dataproc/scenarios/hive_insert/events/run_event.json create mode 100644 producer/spark_dataproc/scenarios/hive_insert/events/schema.json create mode 100644 producer/spark_dataproc/scenarios/hive_insert/events/spark_properties.json create mode 100644 producer/spark_dataproc/scenarios/hive_insert/events/sql.json create mode 100644 producer/spark_dataproc/scenarios/hive_insert/events/symlinks.json create mode 100644 producer/spark_dataproc/scenarios/hive_insert/maintainers.json create mode 100644 producer/spark_dataproc/scenarios/hive_insert/scenario.md create mode 100644 producer/spark_dataproc/scenarios/hive_insert/test/test.py diff --git a/producer/spark_dataproc/scenarios/hive/config.json b/producer/spark_dataproc/scenarios/hive/config.json index 9ef7aecd..f3fab5c7 100644 --- a/producer/spark_dataproc/scenarios/hive/config.json +++ b/producer/spark_dataproc/scenarios/hive/config.json @@ -4,7 +4,7 @@ "max": "3.5.1" }, "openlineage_versions": { - "min": "1.0.0", + "min": "1.55.0", "max": "5.0.0" }, "unchecked_facets": { diff --git a/producer/spark_dataproc/scenarios/hive_insert/config.json b/producer/spark_dataproc/scenarios/hive_insert/config.json new file mode 100644 index 00000000..fa8e145c --- /dev/null +++ b/producer/spark_dataproc/scenarios/hive_insert/config.json @@ -0,0 +1,137 @@ +{ + "component_versions": { + "min": "3.1.0", + "max": "3.5.1" + }, + "openlineage_versions": { + "min": "1.0.0", + "max": "5.0.0" + }, + "unchecked_facets": { + "catalog": [ + "1.38.0", + "1.39.0", + "1.40.0", + "1.40.1" + ] + }, + "tests": [ + { + "name": "run_event_test", + "path": "events/run_event.json", + "tags": { + "facets": [ + "run_event" + ] + } + }, + { + "name": "parent_test", + "path": "events/parent.json", + "tags": { + "facets": [ + "parent" + ] + } + }, + { + "name": "spark_properties_test", + "path": "events/spark_properties.json", + "tags": { + "facets": [ + "spark_properties" + ] + } + }, + { + "name": "processing_engine_test", + "path": "events/processing_engine.json", + "tags": { + "facets": [ + "processing_engine" + ] + } + }, + { + "name": "gcp_dataproc_spark_test", + "path": "events/gcp_dataproc_spark.json", + "openlineage_versions": { + "max": "1.23.0" + }, + "tags": { + "facets": [ + "gcp_dataproc_spark" + ] + } + }, + { + "name": "gcp_dataproc_test", + "path": "events/gcp_dataproc.json", + "openlineage_versions": { + "min": "1.24.0" + }, + "tags": { + "facets": [ + "gcp_dataproc" + ] + } + }, + { + "name": "jobType_test", + "path": "events/jobType.json", + "tags": { + "facets": [ + "jobType" + ] + } + }, + { + "name": "gcp_lineage_test", + "path": "events/gcp_lineage.json", + "tags": { + "facets": [ + "gcp_lineage" + ] + } + }, + { + "name": "dataSource_test", + "path": "events/dataSource.json", + "tags": { + "facets": [ + "dataSource" + ] + } + }, + { + "name": "schema_test", + "path": "events/schema.json", + "tags": { + "facets": [ + "schema" + ], + "lineage_level": { + "hive": [ + "dataset" + ] + } + } + }, + { + "name": "columnLineage_test", + "path": "events/columnLineage.json", + "tags": { + "facets": [ + "columnLineage" + ], + "lineage_level": { + "hive": [ + "dataset", + "column", + "transformation" + ] + } + } + } + ] +} diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/columnLineage.json b/producer/spark_dataproc/scenarios/hive_insert/events/columnLineage.json new file mode 100644 index 00000000..90bff91c --- /dev/null +++ b/producer/spark_dataproc/scenarios/hive_insert/events/columnLineage.json @@ -0,0 +1,53 @@ +{ + "eventType": "COMPLETE", + "job": { + "namespace": "default", + "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + }, + "outputs": [ + { + "namespace": "{{ any(result) }}", + "name": "/user/hive/warehouse/t2", + "facets": { + "columnLineage": { + "fields": { + "a": { + "inputFields": [ + { + "namespace": "{{ any(result) }}", + "name": "/user/hive/warehouse/t1", + "field": "a", + "transformations": [ + { + "type": "DIRECT", + "subtype": "IDENTITY", + "description": "", + "masking": false + } + ] + } + ] + }, + "b": { + "inputFields": [ + { + "namespace": "{{ any(result) }}", + "name": "/user/hive/warehouse/t1", + "field": "b", + "transformations": [ + { + "type": "DIRECT", + "subtype": "IDENTITY", + "description": "", + "masking": false + } + ] + } + ] + } + } + } + } + } + ] +} \ No newline at end of file diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/dataSource.json b/producer/spark_dataproc/scenarios/hive_insert/events/dataSource.json new file mode 100644 index 00000000..83da03c2 --- /dev/null +++ b/producer/spark_dataproc/scenarios/hive_insert/events/dataSource.json @@ -0,0 +1,31 @@ +{ + "eventType": "COMPLETE", + "job": { + "namespace": "default", + "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + }, + "inputs": [ + { + "namespace": "{{ any(result) }}", + "name": "/user/hive/warehouse/t1", + "facets": { + "dataSource": { + "name": "{{ any(result) }}", + "uri": "{{ any(result) }}" + } + } + } + ], + "outputs": [ + { + "namespace": "{{ any(result) }}", + "name": "/user/hive/warehouse/t2", + "facets": { + "dataSource": { + "name": "{{ any(result) }}", + "uri": "{{ any(result) }}" + } + } + } + ] +} \ No newline at end of file diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/gcp_dataproc.json b/producer/spark_dataproc/scenarios/hive_insert/events/gcp_dataproc.json new file mode 100644 index 00000000..91423db1 --- /dev/null +++ b/producer/spark_dataproc/scenarios/hive_insert/events/gcp_dataproc.json @@ -0,0 +1,17 @@ +{ + "eventType": "COMPLETE", + "job": { + "namespace": "default", + "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + }, + "run": { + "facets": { + "gcp_dataproc": { + "queryNodeName": "execute_insert_into_hive_table", + "appName": "hive insert test", + "appId": "{{ any(result) }}", + "projectId": "gcp-open-lineage-testing" + } + } + } +} \ No newline at end of file diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/gcp_dataproc_spark.json b/producer/spark_dataproc/scenarios/hive_insert/events/gcp_dataproc_spark.json new file mode 100644 index 00000000..b7c81eb4 --- /dev/null +++ b/producer/spark_dataproc/scenarios/hive_insert/events/gcp_dataproc_spark.json @@ -0,0 +1,17 @@ +{ + "eventType": "COMPLETE", + "job": { + "namespace": "default", + "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + }, + "run": { + "facets": { + "gcp_dataproc_spark": { + "queryNodeName": "execute_insert_into_hive_table", + "appName": "hive insert test", + "appId": "{{ any(result) }}", + "projectId": "gcp-open-lineage-testing" + } + } + } +} \ No newline at end of file diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/gcp_lineage.json b/producer/spark_dataproc/scenarios/hive_insert/events/gcp_lineage.json new file mode 100644 index 00000000..bb6043e5 --- /dev/null +++ b/producer/spark_dataproc/scenarios/hive_insert/events/gcp_lineage.json @@ -0,0 +1,15 @@ +{ + "eventType": "COMPLETE", + "job": { + "namespace": "default", + "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2", + "facets": { + "gcp_lineage": { + "origin": { + "sourceType": "DATAPROC", + "name": "projects/gcp-open-lineage-testing/regions/us-west1/unknown/" + } + } + } + } +} \ No newline at end of file diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/jobType.json b/producer/spark_dataproc/scenarios/hive_insert/events/jobType.json new file mode 100644 index 00000000..68792d12 --- /dev/null +++ b/producer/spark_dataproc/scenarios/hive_insert/events/jobType.json @@ -0,0 +1,14 @@ +{ + "eventType": "COMPLETE", + "job": { + "namespace": "default", + "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2", + "facets": { + "jobType": { + "processingType": "BATCH", + "integration": "SPARK", + "jobType": "SQL_JOB" + } + } + } +} \ No newline at end of file diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/parent.json b/producer/spark_dataproc/scenarios/hive_insert/events/parent.json new file mode 100644 index 00000000..f3d17a01 --- /dev/null +++ b/producer/spark_dataproc/scenarios/hive_insert/events/parent.json @@ -0,0 +1,20 @@ +{ + "eventType": "COMPLETE", + "job": { + "namespace": "default", + "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + }, + "run": { + "facets": { + "parent": { + "run": { + "runId": "{{ is_uuid(result) }}" + }, + "job": { + "namespace": "default", + "name": "hive_insert_test" + } + } + } + } +} \ No newline at end of file diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/processing_engine.json b/producer/spark_dataproc/scenarios/hive_insert/events/processing_engine.json new file mode 100644 index 00000000..02f08597 --- /dev/null +++ b/producer/spark_dataproc/scenarios/hive_insert/events/processing_engine.json @@ -0,0 +1,16 @@ +{ + "eventType": "COMPLETE", + "job": { + "namespace": "default", + "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + }, + "run": { + "facets": { + "processing_engine": { + "version": "{{ any(result) }}", + "name": "spark", + "openlineageAdapterVersion": "{{ any(result) }}" + } + } + } +} \ No newline at end of file diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/run_event.json b/producer/spark_dataproc/scenarios/hive_insert/events/run_event.json new file mode 100644 index 00000000..4628b28d --- /dev/null +++ b/producer/spark_dataproc/scenarios/hive_insert/events/run_event.json @@ -0,0 +1,28 @@ +{ + "eventType": "COMPLETE", + "run": { + "runId": "{{ is_uuid(result) }}", + "facets": {} + }, + "job": { + "namespace": "default", + "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2", + "facets": {} + }, + "inputs": [ + { + "namespace": "{{ any(result) }}", + "name": "/user/hive/warehouse/t1", + "facets": {}, + "inputFacets": {} + } + ], + "outputs": [ + { + "namespace": "{{ any(result) }}", + "name": "/user/hive/warehouse/t2", + "facets": {}, + "outputFacets": {} + } + ] +} \ No newline at end of file diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/schema.json b/producer/spark_dataproc/scenarios/hive_insert/events/schema.json new file mode 100644 index 00000000..1754e923 --- /dev/null +++ b/producer/spark_dataproc/scenarios/hive_insert/events/schema.json @@ -0,0 +1,47 @@ +{ + "eventType": "COMPLETE", + "job": { + "namespace": "default", + "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + }, + "inputs": [ + { + "namespace": "{{ any(result) }}", + "name": "/user/hive/warehouse/t1", + "facets": { + "schema": { + "fields": [ + { + "name": "a", + "type": "integer" + }, + { + "name": "b", + "type": "string" + } + ] + } + } + } + ], + "outputs": [ + { + "namespace": "{{ any(result) }}", + "name": "/user/hive/warehouse/t2", + "facets": { + "schema": { + "fields": [ + { + "name": "a", + "type": "integer" + }, + { + "name": "b", + "type": "string" + } + ] + } + } + } + ] +} \ No newline at end of file diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/spark_properties.json b/producer/spark_dataproc/scenarios/hive_insert/events/spark_properties.json new file mode 100644 index 00000000..75e12510 --- /dev/null +++ b/producer/spark_dataproc/scenarios/hive_insert/events/spark_properties.json @@ -0,0 +1,17 @@ +{ + "eventType": "COMPLETE", + "job": { + "namespace": "default", + "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + }, + "run": { + "facets": { + "spark_properties": { + "properties": { + "spark.master": "local", + "spark.app.name": "hive insert test" + } + } + } + } +} \ No newline at end of file diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/sql.json b/producer/spark_dataproc/scenarios/hive_insert/events/sql.json new file mode 100644 index 00000000..0112a593 --- /dev/null +++ b/producer/spark_dataproc/scenarios/hive_insert/events/sql.json @@ -0,0 +1,12 @@ +{ + "eventType": "COMPLETE", + "job": { + "namespace": "default", + "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2", + "facets": { + "sql": { + "query": "INSERT INTO t2 SELECT * FROM t1;" + } + } + } +} \ No newline at end of file diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/symlinks.json b/producer/spark_dataproc/scenarios/hive_insert/events/symlinks.json new file mode 100644 index 00000000..c536a89d --- /dev/null +++ b/producer/spark_dataproc/scenarios/hive_insert/events/symlinks.json @@ -0,0 +1,41 @@ +{ + "eventType": "COMPLETE", + "job": { + "namespace": "default", + "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + }, + "inputs": [ + { + "namespace": "{{ any(result) }}", + "name": "/user/hive/warehouse/t1", + "facets": { + "symlinks": { + "identifiers": [ + { + "namespace": "{{ any(result) }}", + "name": "default.t1", + "type": "TABLE" + } + ] + } + } + } + ], + "outputs": [ + { + "namespace": "{{ any(result) }}", + "name": "/user/hive/warehouse/t2", + "facets": { + "symlinks": { + "identifiers": [ + { + "namespace": "{{ any(result) }}", + "name": "default.t2", + "type": "TABLE" + } + ] + } + } + } + ] +} \ No newline at end of file diff --git a/producer/spark_dataproc/scenarios/hive_insert/maintainers.json b/producer/spark_dataproc/scenarios/hive_insert/maintainers.json new file mode 100644 index 00000000..d68b35be --- /dev/null +++ b/producer/spark_dataproc/scenarios/hive_insert/maintainers.json @@ -0,0 +1,14 @@ +[ + { + "type": "maintainer", + "github-name": "ddebowczyk92", + "email": "dominik.debowczyk@getindata.com", + "link": "" + }, + { + "type": "maintainer", + "github-name": "tnazarew", + "email": "tomasz.nazarewicz@getindata.com", + "link": "" + } +] \ No newline at end of file diff --git a/producer/spark_dataproc/scenarios/hive_insert/scenario.md b/producer/spark_dataproc/scenarios/hive_insert/scenario.md new file mode 100644 index 00000000..26d48ea6 --- /dev/null +++ b/producer/spark_dataproc/scenarios/hive_insert/scenario.md @@ -0,0 +1,34 @@ +# Description +Scenario contains a spark job that executes spark sql queries + +DROP TABLE IF EXISTS t1; +DROP TABLE IF EXISTS t2; + +CREATE TABLE IF NOT EXISTS t1 (a INT, b STRING); + +INSERT INTO t1 VALUES (1,2),(3,4); +CREATE TABLE IF NOT EXISTS t2 (a INT, b STRING); +INSERT INTO t2 SELECT * FROM t1; + +It covers the same lineage as the `hive` scenario, without CTAS. CTAS runs a nested insert whose COMPLETE +event often has the same millisecond `eventTime` as the CTAS COMPLETE, and `GcsTransport` names files by +that timestamp, so one of the events is dropped. The `hive` scenario is enabled again once the transport is fixed. + +# Entities + +input entity is hive table + +`default.t1` + +output entity hive table +`default.t2` + +# Facets +Facets present in the events: +- ColumnLineageDatasetFacet +- DatasourceDatasetFacet +- JobTypeJobFacet +- ParentRunFacet +- SQLJobFacet +- SchemaDatasetFacet +- SymlinksDatasetFacet diff --git a/producer/spark_dataproc/scenarios/hive_insert/test/test.py b/producer/spark_dataproc/scenarios/hive_insert/test/test.py new file mode 100644 index 00000000..bb7e9b07 --- /dev/null +++ b/producer/spark_dataproc/scenarios/hive_insert/test/test.py @@ -0,0 +1,19 @@ +import time + +from pyspark.sql import SparkSession + +spark = SparkSession.builder.master("local").appName("hive insert test").enableHiveSupport().getOrCreate() + +spark.sql("DROP TABLE IF EXISTS t1;") +spark.sql("DROP TABLE IF EXISTS t2;") + +spark.sql("CREATE TABLE IF NOT EXISTS t1 (a INT, b STRING);") +spark.sql("INSERT INTO t1 VALUES (1,2),(3,4);") +# CREATE + INSERT instead of CTAS: CTAS runs a nested insert whose COMPLETE event often has the same +# millisecond eventTime as the CTAS COMPLETE, and GcsTransport drops one of them (see hive scenario) +spark.sql("CREATE TABLE IF NOT EXISTS t2 (a INT, b STRING);") +spark.sql("INSERT INTO t2 SELECT * FROM t1;") + +time.sleep(3) + +spark.sql("SELECT * FROM t2;").show()