rfc: spark structured streaming sink and source platform

active/000-spark-streaming-sink-source-platform.md

@@ -0,0 +1,119 @@
+- Start Date: 2025-04-21
+- RFC PR: [#11](https://github.com/datahub-project/rfcs/pull/11)
+- Discussion Issue: (GitHub issue this was discussed in before the RFC, if any)
+- Implementation PR(s): (leave this empty)
+
+# Spark Streaming Sink/Source Platform
+
+## Summary
+
+Allows configuration of the data platform for Spark Structured Streaming sources and sinks.
+
+## Motivation
+
+The motivation for this RFC stems from an issue encountered while capturing data lineage using DataHub with Spark Structured Streaming. In the DataHub [code](https://github.com/datahub-project/datahub/blob/master/metadata-integration/java/acryl-spark-lineage/src/main/java/datahub/spark/converter/SparkStreamingEventToDatahub.java#L145), a regular expression matcher expects sources to be prefixed with identifiers like Kafka[…] to determine the data platform. However, since Iceberg tables lack such a prefix (e.g., iceberg[…]), DataHub fails to recognize the platform and thus shows no lineage.
+
+## Requirements
+
+- The proposal should be able to identify the data platform of a source or sink based on the configuration provided in the Spark job.
+
+## Detailed design
+
+It is proposed to introduce two new Spark configurations: `spark.datahub.streaming_platform` for specifying the streaming platform, and alias‐based streaming configuration (see *Configuring Iceberg-based dataset URNs* below). Within the `generateUrnFromStreamingDescription` method in `SparkStreamingEventToDatahub.java`, these configurations will serve as fallbacks in cases where the regular expression matcher fails to extract the platform. If the configurations are set, their values will be used to determine the data platform. An example implementation is shown below:
+```java
+  public static Optional<DatasetUrn> generateUrnFromStreamingDescription(
+        String description, SparkLineageConf sparkLineageConf) {
+    return SparkStreamingEventToDatahub.generateUrnFromStreamingDescription(
+            description, sparkLineageConf, false);
+}
+```
+```java
+  public static Optional<DatasetUrn> generateUrnFromStreamingDescription(String description,
+                                                                         SparkLineageConf sparkLineageConf, boolean isSink) {
+    String pattern = "(.*?)\\[(.*)]";
+    Pattern r = Pattern.compile(pattern);
+    Matcher m = r.matcher(description);
+    if (m.find()) {
+        String namespace = m.group(1);
+        String platform = getDatahubPlatform(namespace);
+        String path = m.group(2);
+        log.debug("Streaming description Platform: {}, Path: {}", platform, path);
+        if (platform.equals(KAFKA_PLATFORM)) {
+            path = getKafkaTopicFromPath(m.group(2));
+        } else if (platform.equals(FILE_PLATFORM) || platform.equals(DELTA_LAKE_PLATFORM)) {
+            try {
+                DatasetUrn urn = HdfsPathDataset.create(new URI(path), sparkLineageConf.getOpenLineageConf()).urn();
+                return Optional.of(urn);
+            } catch (InstantiationException e) {
+                return Optional.empty();
+            } catch (URISyntaxException e) {
+                log.error("Failed to parse path {}", path, e);
+                return Optional.empty();
+            }
+        }
+        return Optional.of(
+                new DatasetUrn(new DataPlatformUrn(platform), path, sparkLineageConf.getOpenLineageConf().getFabricType()));
+    } else {
+        if (sparkLineageConf.getOpenLineageConf().getStreamingPlatform() != null) {
+            try {
+                CatalogTableDataset catalogTableDataset =
+                        CatalogTableDataset.create(sparkLineageConf.getOpenLineageConf().getStreamingPlatform(), description,
+                                sparkLineageConf.getOpenLineageConf(), isSink ? "sink" : "source");
+                if (catalogTableDataset == null) {
+                    return Optional.empty();
+                } else {
+                    DatasetUrn urn = catalogTableDataset.urn();
+                    return Optional.of(urn);
+                }
+            } catch (InstantiationException e) {
+                return Optional.empty();
+            }
+        } else {
+            return Optional.empty();
+        }
+    }
+}
+```
+### Configuring Iceberg-based dataset URNs
+
+This section follows the approach described in [Configuring Hdfs based dataset URNs](https://datahubproject.io/docs/metadata-integration/java/acryl-spark-lineage/#configuring-hdfs-based-dataset-urns)
+
+Spark emits lineage between datasets. It has its own logic for generating urns. Python sources emit metadata of
+datasets. To link these 2 things, urns generated by both have to match.
+This section will help you to match urns to that of other ingestion sources.
+By default, URNs are created using
+template `urn:li:dataset:(urn:li:dataPlatform:<$platform>,<$platformInstance>.<$name>,<$env>)`. We can configure these 4
+things to generate the desired urn.
+
+**Platform**:
+The platform is explicitly supported through the new Spark configuration key:
+
+- `spark.datahub.streaming_platform`
+
+Platforms that do not set this configuration will default to `null`.
+
+**Name**:
+By default, the name is the complete path.
+
+**platform instance and env:**
+
+The default value for env is 'PROD' and the platform instance is None. env and platform instances can be set for all
+datasets using configurations `spark.datahub.streaming.platform.<$platform>.<streaming_alias>.platformInstance` and `spark.datahub.streaming.platform.<$platform>.<streaming_alias>.env`.
+If spark is processing data that belongs to a different env or platform instance, then 'streaming_alias' can be used to
+specify `streaming_spec` specific values of these. 'streaming_alias' groups the env and platform instance
+together.
+
+streaming_alias_list Example:
+
+The below example explains the configuration of the case, where data from 2 Iceberg tables are being processed in a single
+spark application and data from my_table_1 are supposed to have "instance1" as platform instance and "PROD" as env, and
+data from my_table_2 should have env "DEV" in their dataset URNs.
+
+```properties
+spark.datahub.streaming.platform.iceberg.stream1.env : PROD
+spark.datahub.streaming.platform.iceberg.stream1.streaming_io_platform_type : source
+spark.datahub.streaming.platform.iceberg.stream1.platformInstance : instance1
+spark.datahub.streaming.platform.iceberg.stream2.env : DEV
+spark.datahub.streaming.platform.iceberg.stream2.streaming_io_platform_type : sink
+spark.datahub.streaming.platform.iceberg.stream2.platformInstance : instance2
+```