Implementation:Datahub project Datahub StreamingDataSourceV2RelationVisitor

Knowledge Sources	Datahub_project_Datahub
Domains	Spark_Lineage, OpenLineage
Last Updated	2026-02-10 00:00 GMT

Overview

StreamingDataSourceV2RelationVisitor is a query plan visitor in the io.openlineage.spark.agent.lifecycle.plan package that extracts input datasets from Spark Structured Streaming logical plan nodes of type StreamingDataSourceV2Relation. It extends QueryPlanVisitor<StreamingDataSourceV2Relation, InputDataset> and uses a strategy pattern to select the appropriate stream-specific handler based on the runtime class of the streaming micro-batch source.

The visitor supports Kafka, Kinesis, MongoDB, and file-based streaming sources out of the box, and falls back to a no-op strategy for unrecognized stream types.

Source file: metadata-integration/java/acryl-spark-lineage/src/main/java/io/openlineage/spark/agent/lifecycle/plan/StreamingDataSourceV2RelationVisitor.java (113 lines)

Code Reference

Class Declaration

@Slf4j
public class StreamingDataSourceV2RelationVisitor
    extends QueryPlanVisitor<StreamingDataSourceV2Relation, InputDataset> {

Supported Stream Class Constants

private static final String KAFKA_MICRO_BATCH_STREAM_CLASS_NAME =
    "org.apache.spark.sql.kafka010.KafkaMicroBatchStream";
private static final String KINESIS_MICRO_BATCH_STREAM_CLASS_NAME =
    "org.apache.spark.sql.connector.kinesis.KinesisV2MicrobatchStream";
private static final String MONGO_MICRO_BATCH_STREAM_CLASS_NAME =
    "com.mongodb.spark.sql.connector.read.MongoMicroBatchStream";
private static final String FILE_STREAM_MICRO_BATCH_STREAM_CLASS_NAME =
    "org.apache.spark.sql.execution.streaming.sources.FileStreamSourceV2";

Constructor

public StreamingDataSourceV2RelationVisitor(@NonNull OpenLineageContext context)

Key Methods

isDefinedAt

@Override
public boolean isDefinedAt(LogicalPlan x)

Returns true if the logical plan node is an instance of StreamingDataSourceV2Relation.

apply

@Override
public List<InputDataset> apply(LogicalPlan x)

Casts the logical plan to StreamingDataSourceV2Relation, selects a stream strategy via selectStrategy, and returns the input datasets produced by that strategy.

selectStrategy

public StreamStrategy selectStrategy(StreamingDataSourceV2Relation relation)

Inspects the canonical class name of relation.stream() and selects one of:

• KafkaMicroBatchStreamStrategy - for Kafka sources
• KinesisMicroBatchStreamStrategy - for Kinesis sources
• MongoMicroBatchStreamStrategy - for MongoDB sources
• FileStreamMicroBatchStreamStrategy - for file-based sources (including Parquet, JSON, CSV variants)
• NoOpStreamStrategy - fallback for unrecognized stream types

isFileBasedStreamingSource

private boolean isFileBasedStreamingSource(String streamClassName)

Heuristic check that returns true if the class name contains substrings indicating a file-based streaming source, covering FileStreamSource, TextFileStreamSource, FileSource, ParquetFileSource, JsonFileSource, CsvFileSource, and standard Spark datasources v2 packages for CSV, JSON, and Parquet.

I/O Contract

Direction	Type	Description
Input	LogicalPlan (specifically StreamingDataSourceV2Relation)	A Spark logical plan node representing a streaming data source relation.
Input	OpenLineageContext	Context carrying the OpenLineage client, Spark context, and dataset factory methods.
Output	List<InputDataset>	Zero or more OpenLineage InputDataset objects representing the streaming sources (e.g., Kafka topics, Kinesis streams, MongoDB collections, file paths).

Usage Examples

This visitor is registered with the OpenLineage plan traversal framework and is invoked automatically when the query plan walker encounters a StreamingDataSourceV2Relation node:

// Registered as part of the visitor list in the OpenLineage Spark integration
List<QueryPlanVisitor<?, ? extends InputDataset>> inputVisitors = Arrays.asList(
    new StreamingDataSourceV2RelationVisitor(context),
    // ... other visitors
);

// During plan traversal, the framework calls:
// visitor.isDefinedAt(logicalPlan) -> true if StreamingDataSourceV2Relation
// visitor.apply(logicalPlan) -> returns List<InputDataset>

Related Pages

• Datahub_project_Datahub_WriteToDataSourceV2Visitor - Companion visitor handling output datasets for streaming writes
• Datahub_project_Datahub_SparkStreamingEventToDatahub - Higher-level converter that produces DataHub MCPs from streaming events
• Datahub_project_Datahub_SparkPathUtils - Path resolution utilities used by file-based stream strategies
• Datahub_project_Datahub_RemovePathPatternUtils - Path pattern cleanup applied to extracted dataset names