From cb1d9f134906c2bbdab9622ccd0e51f0c25e6b58 Mon Sep 17 00:00:00 2001 From: tnazarew Date: Fri, 16 Oct 2026 16:05:28 +0200 Subject: [PATCH 1/4] 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() From 09ca918016b0f9670dc4b3db347124e196b4d6ba Mon Sep 17 00:00:00 2001 From: tnazarew Date: Wed, 21 Oct 2026 18:27:27 +0200 Subject: [PATCH 2/4] hive_insert: remove ddebowczyk92 from maintainers Co-Authored-By: Claude Opus 5.5 Signed-off-by: tnazarew --- .../spark_dataproc/scenarios/hive_insert/maintainers.json | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/producer/spark_dataproc/scenarios/hive_insert/maintainers.json b/producer/spark_dataproc/scenarios/hive_insert/maintainers.json index d68b35be..fadb1ecd 100644 --- a/producer/spark_dataproc/scenarios/hive_insert/maintainers.json +++ b/producer/spark_dataproc/scenarios/hive_insert/maintainers.json @@ -1,14 +1,8 @@ [ - { - "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 +] From cfb5829589272ca647c310b03041a77760be84a6 Mon Sep 17 00:00:00 2001 From: tnazarew Date: Wed, 21 Oct 2026 23:34:54 +0200 Subject: [PATCH 3/4] hive_insert: use a database named after the test Since #344 one cluster runs all scenarios of an OpenLineage version, up to 3 at a time, and hive_insert uses the same t1/t2 tables as hive. Like hive, it now creates and uses the database from spark.test.namespace (the scenario name). Expected dataset and job names follow the names the hive scenario produced in CI: /tmp/warehouse/.db/, .
symlinks, and _db_t2 in the insert job name. Co-Authored-By: Claude Opus 5.5 Signed-off-by: tnazarew --- .../scenarios/hive_insert/events/columnLineage.json | 8 ++++---- .../scenarios/hive_insert/events/dataSource.json | 6 +++--- .../scenarios/hive_insert/events/gcp_dataproc.json | 2 +- .../hive_insert/events/gcp_dataproc_spark.json | 2 +- .../scenarios/hive_insert/events/gcp_lineage.json | 2 +- .../scenarios/hive_insert/events/jobType.json | 2 +- .../scenarios/hive_insert/events/parent.json | 2 +- .../hive_insert/events/processing_engine.json | 2 +- .../scenarios/hive_insert/events/run_event.json | 6 +++--- .../scenarios/hive_insert/events/schema.json | 6 +++--- .../scenarios/hive_insert/events/spark_properties.json | 2 +- .../scenarios/hive_insert/events/sql.json | 2 +- .../scenarios/hive_insert/events/symlinks.json | 10 +++++----- .../spark_dataproc/scenarios/hive_insert/test/test.py | 5 +++++ 14 files changed, 31 insertions(+), 26 deletions(-) diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/columnLineage.json b/producer/spark_dataproc/scenarios/hive_insert/events/columnLineage.json index 90bff91c..466fc45e 100644 --- a/producer/spark_dataproc/scenarios/hive_insert/events/columnLineage.json +++ b/producer/spark_dataproc/scenarios/hive_insert/events/columnLineage.json @@ -2,12 +2,12 @@ "eventType": "COMPLETE", "job": { "namespace": "default", - "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + "name": "hive_insert_test.execute_insert_into_hive_table.hive_insert_db_t2" }, "outputs": [ { "namespace": "{{ any(result) }}", - "name": "/user/hive/warehouse/t2", + "name": "/tmp/warehouse/hive_insert.db/t2", "facets": { "columnLineage": { "fields": { @@ -15,7 +15,7 @@ "inputFields": [ { "namespace": "{{ any(result) }}", - "name": "/user/hive/warehouse/t1", + "name": "/tmp/warehouse/hive_insert.db/t1", "field": "a", "transformations": [ { @@ -32,7 +32,7 @@ "inputFields": [ { "namespace": "{{ any(result) }}", - "name": "/user/hive/warehouse/t1", + "name": "/tmp/warehouse/hive_insert.db/t1", "field": "b", "transformations": [ { diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/dataSource.json b/producer/spark_dataproc/scenarios/hive_insert/events/dataSource.json index 83da03c2..259b9b46 100644 --- a/producer/spark_dataproc/scenarios/hive_insert/events/dataSource.json +++ b/producer/spark_dataproc/scenarios/hive_insert/events/dataSource.json @@ -2,12 +2,12 @@ "eventType": "COMPLETE", "job": { "namespace": "default", - "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + "name": "hive_insert_test.execute_insert_into_hive_table.hive_insert_db_t2" }, "inputs": [ { "namespace": "{{ any(result) }}", - "name": "/user/hive/warehouse/t1", + "name": "/tmp/warehouse/hive_insert.db/t1", "facets": { "dataSource": { "name": "{{ any(result) }}", @@ -19,7 +19,7 @@ "outputs": [ { "namespace": "{{ any(result) }}", - "name": "/user/hive/warehouse/t2", + "name": "/tmp/warehouse/hive_insert.db/t2", "facets": { "dataSource": { "name": "{{ any(result) }}", diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/gcp_dataproc.json b/producer/spark_dataproc/scenarios/hive_insert/events/gcp_dataproc.json index 91423db1..db4ae4e1 100644 --- a/producer/spark_dataproc/scenarios/hive_insert/events/gcp_dataproc.json +++ b/producer/spark_dataproc/scenarios/hive_insert/events/gcp_dataproc.json @@ -2,7 +2,7 @@ "eventType": "COMPLETE", "job": { "namespace": "default", - "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + "name": "hive_insert_test.execute_insert_into_hive_table.hive_insert_db_t2" }, "run": { "facets": { 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 index b7c81eb4..a9124fd4 100644 --- a/producer/spark_dataproc/scenarios/hive_insert/events/gcp_dataproc_spark.json +++ b/producer/spark_dataproc/scenarios/hive_insert/events/gcp_dataproc_spark.json @@ -2,7 +2,7 @@ "eventType": "COMPLETE", "job": { "namespace": "default", - "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + "name": "hive_insert_test.execute_insert_into_hive_table.hive_insert_db_t2" }, "run": { "facets": { diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/gcp_lineage.json b/producer/spark_dataproc/scenarios/hive_insert/events/gcp_lineage.json index bb6043e5..7bf6093f 100644 --- a/producer/spark_dataproc/scenarios/hive_insert/events/gcp_lineage.json +++ b/producer/spark_dataproc/scenarios/hive_insert/events/gcp_lineage.json @@ -2,7 +2,7 @@ "eventType": "COMPLETE", "job": { "namespace": "default", - "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2", + "name": "hive_insert_test.execute_insert_into_hive_table.hive_insert_db_t2", "facets": { "gcp_lineage": { "origin": { diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/jobType.json b/producer/spark_dataproc/scenarios/hive_insert/events/jobType.json index 68792d12..d9774086 100644 --- a/producer/spark_dataproc/scenarios/hive_insert/events/jobType.json +++ b/producer/spark_dataproc/scenarios/hive_insert/events/jobType.json @@ -2,7 +2,7 @@ "eventType": "COMPLETE", "job": { "namespace": "default", - "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2", + "name": "hive_insert_test.execute_insert_into_hive_table.hive_insert_db_t2", "facets": { "jobType": { "processingType": "BATCH", diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/parent.json b/producer/spark_dataproc/scenarios/hive_insert/events/parent.json index f3d17a01..197a2eb3 100644 --- a/producer/spark_dataproc/scenarios/hive_insert/events/parent.json +++ b/producer/spark_dataproc/scenarios/hive_insert/events/parent.json @@ -2,7 +2,7 @@ "eventType": "COMPLETE", "job": { "namespace": "default", - "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + "name": "hive_insert_test.execute_insert_into_hive_table.hive_insert_db_t2" }, "run": { "facets": { diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/processing_engine.json b/producer/spark_dataproc/scenarios/hive_insert/events/processing_engine.json index 02f08597..0418e878 100644 --- a/producer/spark_dataproc/scenarios/hive_insert/events/processing_engine.json +++ b/producer/spark_dataproc/scenarios/hive_insert/events/processing_engine.json @@ -2,7 +2,7 @@ "eventType": "COMPLETE", "job": { "namespace": "default", - "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + "name": "hive_insert_test.execute_insert_into_hive_table.hive_insert_db_t2" }, "run": { "facets": { diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/run_event.json b/producer/spark_dataproc/scenarios/hive_insert/events/run_event.json index 4628b28d..4074f286 100644 --- a/producer/spark_dataproc/scenarios/hive_insert/events/run_event.json +++ b/producer/spark_dataproc/scenarios/hive_insert/events/run_event.json @@ -6,13 +6,13 @@ }, "job": { "namespace": "default", - "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2", + "name": "hive_insert_test.execute_insert_into_hive_table.hive_insert_db_t2", "facets": {} }, "inputs": [ { "namespace": "{{ any(result) }}", - "name": "/user/hive/warehouse/t1", + "name": "/tmp/warehouse/hive_insert.db/t1", "facets": {}, "inputFacets": {} } @@ -20,7 +20,7 @@ "outputs": [ { "namespace": "{{ any(result) }}", - "name": "/user/hive/warehouse/t2", + "name": "/tmp/warehouse/hive_insert.db/t2", "facets": {}, "outputFacets": {} } diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/schema.json b/producer/spark_dataproc/scenarios/hive_insert/events/schema.json index 1754e923..fb062c21 100644 --- a/producer/spark_dataproc/scenarios/hive_insert/events/schema.json +++ b/producer/spark_dataproc/scenarios/hive_insert/events/schema.json @@ -2,12 +2,12 @@ "eventType": "COMPLETE", "job": { "namespace": "default", - "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + "name": "hive_insert_test.execute_insert_into_hive_table.hive_insert_db_t2" }, "inputs": [ { "namespace": "{{ any(result) }}", - "name": "/user/hive/warehouse/t1", + "name": "/tmp/warehouse/hive_insert.db/t1", "facets": { "schema": { "fields": [ @@ -27,7 +27,7 @@ "outputs": [ { "namespace": "{{ any(result) }}", - "name": "/user/hive/warehouse/t2", + "name": "/tmp/warehouse/hive_insert.db/t2", "facets": { "schema": { "fields": [ diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/spark_properties.json b/producer/spark_dataproc/scenarios/hive_insert/events/spark_properties.json index 75e12510..3f35ae7f 100644 --- a/producer/spark_dataproc/scenarios/hive_insert/events/spark_properties.json +++ b/producer/spark_dataproc/scenarios/hive_insert/events/spark_properties.json @@ -2,7 +2,7 @@ "eventType": "COMPLETE", "job": { "namespace": "default", - "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + "name": "hive_insert_test.execute_insert_into_hive_table.hive_insert_db_t2" }, "run": { "facets": { diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/sql.json b/producer/spark_dataproc/scenarios/hive_insert/events/sql.json index 0112a593..d374a21d 100644 --- a/producer/spark_dataproc/scenarios/hive_insert/events/sql.json +++ b/producer/spark_dataproc/scenarios/hive_insert/events/sql.json @@ -2,7 +2,7 @@ "eventType": "COMPLETE", "job": { "namespace": "default", - "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2", + "name": "hive_insert_test.execute_insert_into_hive_table.hive_insert_db_t2", "facets": { "sql": { "query": "INSERT INTO t2 SELECT * FROM t1;" diff --git a/producer/spark_dataproc/scenarios/hive_insert/events/symlinks.json b/producer/spark_dataproc/scenarios/hive_insert/events/symlinks.json index c536a89d..63c0a4d3 100644 --- a/producer/spark_dataproc/scenarios/hive_insert/events/symlinks.json +++ b/producer/spark_dataproc/scenarios/hive_insert/events/symlinks.json @@ -2,18 +2,18 @@ "eventType": "COMPLETE", "job": { "namespace": "default", - "name": "hive_insert_test.execute_insert_into_hive_table.warehouse_t2" + "name": "hive_insert_test.execute_insert_into_hive_table.hive_insert_db_t2" }, "inputs": [ { "namespace": "{{ any(result) }}", - "name": "/user/hive/warehouse/t1", + "name": "/tmp/warehouse/hive_insert.db/t1", "facets": { "symlinks": { "identifiers": [ { "namespace": "{{ any(result) }}", - "name": "default.t1", + "name": "hive_insert.t1", "type": "TABLE" } ] @@ -24,13 +24,13 @@ "outputs": [ { "namespace": "{{ any(result) }}", - "name": "/user/hive/warehouse/t2", + "name": "/tmp/warehouse/hive_insert.db/t2", "facets": { "symlinks": { "identifiers": [ { "namespace": "{{ any(result) }}", - "name": "default.t2", + "name": "hive_insert.t2", "type": "TABLE" } ] diff --git a/producer/spark_dataproc/scenarios/hive_insert/test/test.py b/producer/spark_dataproc/scenarios/hive_insert/test/test.py index bb7e9b07..efc55467 100644 --- a/producer/spark_dataproc/scenarios/hive_insert/test/test.py +++ b/producer/spark_dataproc/scenarios/hive_insert/test/test.py @@ -4,6 +4,11 @@ spark = SparkSession.builder.master("local").appName("hive insert test").enableHiveSupport().getOrCreate() +# local datasets live in a database named after the test, so scenarios running at the same time don't collide +namespace = spark.conf.get("spark.test.namespace", "hive_insert") +spark.sql(f"CREATE DATABASE IF NOT EXISTS {namespace};") +spark.sql(f"USE {namespace};") + spark.sql("DROP TABLE IF EXISTS t1;") spark.sql("DROP TABLE IF EXISTS t2;") From 38a13851d683233172b2634eccd7b3b773026e42 Mon Sep 17 00:00:00 2001 From: tnazarew Date: Wed, 21 Oct 2026 23:35:05 +0200 Subject: [PATCH 4/4] hive_insert: update table names in scenario.md Co-Authored-By: Claude Opus 5.5 Signed-off-by: tnazarew --- producer/spark_dataproc/scenarios/hive_insert/scenario.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/producer/spark_dataproc/scenarios/hive_insert/scenario.md b/producer/spark_dataproc/scenarios/hive_insert/scenario.md index 26d48ea6..e6c90427 100644 --- a/producer/spark_dataproc/scenarios/hive_insert/scenario.md +++ b/producer/spark_dataproc/scenarios/hive_insert/scenario.md @@ -18,10 +18,10 @@ that timestamp, so one of the events is dropped. The `hive` scenario is enabled input entity is hive table -`default.t1` +`hive_insert.t1` output entity hive table -`default.t2` +`hive_insert.t2` # Facets Facets present in the events: