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()