[SPARK-24647] Sink Should Return Writen Offsets For ProgressReporting

• Type: Improvement
• Status: Resolved
• Priority: Major
• Resolution: Incomplete
• Affects Version/s: 2.3.1
• Component/s: Structured Streaming
• Labels: bulk-closed
• Assignee: Unassigned
• Reporter: Vaclav Kosar
• Created: 25/Jun/18 12:15
• Updated: 12/Dec/22 18:10
• Resolved: 25/May/21 01:42

Description

To be able to track data lineage for Structured Streaming (I intend to implement this to Open Source Project Spline), the monitoring needs to be able to not only to track where the data was read from but also where results were written to. This could be to my knowledge best implemented using monitoring StreamingQueryProgress. However currently written data offsets are not available on Sink or StreamWriter interface. Implementing as proposed would also bring symmetry to StreamingQueryProgress fields sources and sink.

 

Similar Proposals

Made in following jiras. These would not be sufficient for lineage tracking.

• https://issues.apache.org/jira/browse/SPARK-18258
• https://issues.apache.org/jira/browse/SPARK-21313

 

Current State

• Method Sink#addBatch returns Unit.
• Object WriterCommitMessage does not carry any progress information about comitted rows.
• StreamingQueryProgress reports offsetSeq start and end using sourceProgress value but sinkProgress only calls toString method.

  "sources" : [ {
    "description" : "KafkaSource[Subscribe[test-topic]]",
    "startOffset" : null,
    "endOffset" : { "test-topic" : { "0" : 5000 }},
    "numInputRows" : 5000,
    "processedRowsPerSecond" : 645.3278265358803
  } ],
  "sink" : {
    "description" : "org.apache.spark.sql.execution.streaming.ConsoleSink@9da556f"
  }

 

 

Proposed State

• Implement support only for v2 sinks as those are to use used in future.
• WriterCommitMessage to hold optional min and max offset information of commited rows e.g. Kafka does it by returning RecordMetadata object from send method.
• StreamingQueryProgress incorporate sinkProgress in similar fashion as sourceProgress.

 

 

  "sources" : [ {
    "description" : "KafkaSource[Subscribe[test-topic]]",
    "startOffset" : null,
    "endOffset" : { "test-topic" : { "0" : 5000 }},
    "numInputRows" : 5000,
    "processedRowsPerSecond" : 645.3278265358803
  } ],
  "sink" : {
    "description" : "org.apache.spark.sql.execution.streaming.ConsoleSink@9da556f",
   "startOffset" : null,
    "endOffset" { "sinkTopic": { "0": 333 }}
  }

 

Implementation

• PR submitters: Me and wajda as soon as prerequisite jira is merged.
• Sinks: Modify all v2 sinks to conform a new interface or return dummy values.
• ProgressReporter: Merge offsets from different batches properly, similarly to how it is done for sources.

 

Issue Links

• depends upon: SPARK-24933 SinkProgress should report written rows
• is related to: SPARK-18258 Sinks need access to offset representation
• is related to: SPARK-24748 Support for reporting custom metrics via Streaming Query Progress
• is related to: SPARK-21313 ConsoleSink's string representation
• links to: [Github] Pull Request #22143 (vackosar)
• links to: GitHub Pull Request #22143