[SPARK-22732] Add Structured Streaming APIs to DataSourceV2 #19925

jose-torres · 2017-12-07T22:39:38Z

What changes were proposed in this pull request?

This PR provides DataSourceV2 API support for structured streaming, including new pieces needed to support continuous processing [SPARK-20928]. High level summary:

DataSourceV2 includes new mixins to support micro-batch and continuous reads and writes. For reads, we accept an optional user specified schema rather than using the ReadSupportWithSchema model, because doing so would severely complicate the interface.
DataSourceV2Reader includes new interfaces to read a specific microbatch or read continuously from a given offset. These follow the same setter pattern as the existing Supports* mixins so that they can work with SupportsScanUnsafeRow.
DataReader (the per-partition reader) has a new subinterface ContinuousDataReader only for continuous processing. This reader has a special method to check progress, and next() blocks for new input rather than returning false.
Offset, an abstract representation of position in a streaming query, is ported to the public API. (Each type of reader will define its own Offset implementation.)
DataSourceV2Writer has a new subinterface ContinuousWriter only for continuous processing. Commits to this interface come tagged with an epoch number, as the execution engine will continue to produce new epoch commits as the task continues indefinitely.

Note that this PR does not propose to change the existing DataSourceV2 batch API, or deprecate the existing streaming source/sink internal APIs in spark.sql.execution.streaming.

How was this patch tested?

Toy implementations of the new interfaces with unit tests.

SparkQA · 2017-12-07T22:43:43Z

Test build #84622 has finished for PR 19925 at commit 1fdb2cc.

This patch fails RAT tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-12-07T22:55:37Z

Test build #84623 has finished for PR 19925 at commit be0b8e9.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-12-08T02:13:27Z

Test build #84624 has finished for PR 19925 at commit 22d07cd.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

brkyvz · 2017-12-08T17:47:40Z

sql/core/src/main/java/org/apache/spark/sql/sources/v2/MicroBatchWriteSupport.java

+   *                contain the same records.
+   * @param schema the schema of the data to be written.
+   * @param mode the output mode which determines what successive batch output means to this
+   *             source, please refer to {@link OutputMode} for more details.


to this sink? not source

Good point. Fixed here and in ContinuousWriteSupport.

SparkQA · 2017-12-11T23:28:33Z

Test build #84728 has finished for PR 19925 at commit 7c46b33.

This patch fails SparkR unit tests.
This patch merges cleanly.
This patch adds no public classes.

jose-torres · 2017-12-11T23:35:40Z

retest this please

SparkQA · 2017-12-12T02:10:25Z

Test build #84737 has finished for PR 19925 at commit 7c46b33.

This patch fails SparkR unit tests.
This patch merges cleanly.
This patch adds no public classes.

jose-torres · 2017-12-12T22:27:11Z

retest this please

zsxwing

Overall looks good! This is really exciting. Just left some comments.

zsxwing · 2017-12-12T23:17:11Z

sql/core/src/main/java/org/apache/spark/sql/sources/v2/reader/PartitionOffset.java

+ *
+ * These offsets must be serializable.
+ */
+public interface PartitionOffset {


extends Serializable

zsxwing · 2017-12-12T23:27:33Z

sql/core/src/main/java/org/apache/spark/sql/sources/v2/reader/Offset.java

+
+package org.apache.spark.sql.sources.v2.reader;
+
+public abstract class Offset {


nit: add javadoc

zsxwing · 2017-12-12T23:54:46Z

...n/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousRateStreamSource.scala

+  }
+
+  override def readSchema(): StructType = {
+    StructType(


nit: You can reuse org.apache.spark.sql.execution.streaming.RateSourceProvider.SCHEMA

zsxwing · 2017-12-12T23:57:07Z

...n/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousRateStreamSource.scala

+  override def createReadTasks(): java.util.List[ReadTask[Row]] = {
+    val partitionStartMap = Option(offset.orElse(null)).map {
+      case o: ContinuousRateStreamOffset => o.partitionToStartValue
+      case s: SerializedOffset => Serialization.read[Map[Int, Long]](s.json)


A v2 reader should never see SerializedOffset. Right? The engine is supposed to call deserializeOffset to get the real offset object.

zsxwing · 2017-12-12T23:59:29Z

...n/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousRateStreamSource.scala

+      case s: SerializedOffset => Serialization.read[Map[Int, Long]](s.json)
+      case _ => throw new IllegalArgumentException("invalid offset type for ContinuousRateSource")
+    }
+    if (partitionStartMap.exists(_.keySet.size > numPartitions)) {


_.size != numPartitions . Changing the numPartitions will generate either duplicated data or discontinuous data. Right?

zsxwing · 2017-12-13T00:06:05Z

...n/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousRateStreamSource.scala

+      case _ => throw new IllegalArgumentException("invalid offset type for ContinuousRateSource")
+    }
+    if (partitionStartMap.exists(_.keySet.size > numPartitions)) {
+      throw new IllegalArgumentException("Start offset contained too many partitions.")


Could you make the error message more clear? Such as, The previous run has XXX partitions but the new one has XXX partitions. The "numPartitions" option must not be changed.

zsxwing · 2017-12-13T00:16:19Z

...n/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousRateStreamSource.scala

+    // Set the timestamp for the first time.
+    if (currentRow == null) nextReadTime = System.currentTimeMillis() + 1000
+
+    if (numReadRows == rowsPerSecond) {


The timestamp column will be uneven in the current implementation. Could you just calculate how long to sleep for each row and sleep for each row?

I could, but the unevenness matches the microbatch behavior. Having it sleep for each row would make it significantly harder to write reliable tests.

The microbatch source generates even timestamps. Here is the code:

spark/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/RateSourceProvider.scala

Line 213 in 3c0c2d0

InternalRow(DateTimeUtils.fromMillis(relative + localStartTimeMs), v)

Hm, so it does. I'm not sure why I thought it didn't.

Having said that, I still think even timestamps would pose a serious obstacle to writing reliable tests of the continuous processing engine, which is the primary goal of this implementation. Having a second long window to commit data and shut down the query lets us figure out exactly which rows should have made it to the sink, at least in Spark unit tests where everything is running locally. If the rows come evenly, we'll have a much harder time figuring out what data is supposed to be present.

We discussed this offline, and agreed that:

it's best for the timestamps to be even

it's fine to adjust the rate source tests in the full implementation to only check that at least the expected data is present (but more may have come in).

zsxwing · 2017-12-13T00:26:03Z

...n/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousRateStreamSource.scala

+  override def close(): Unit = {}
+
+  override def getOffset(): PartitionOffset =
+    ContinuousRateStreamPartitionOffset(partitionIndex, currentValue - increment)


currentValue - increment will be wrong if the first next() call is interrupted.

(Your comment made me relalize there was a fencepost error here, which I'll swap some things around to fix.)

getOffset(), like get(), is supposed to be paired with next() in a pseudo-iterator and non-thread-safe manner. It'd be incorrect for a caller to look for getOffset() to be correct after a failed next() call.

zsxwing · 2017-12-13T00:30:51Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/memoryV2.scala

+  override def commit(epochId: Long, messages: Array[WriterCommitMessage]): Unit = {
+    val newRows = messages.flatMap {
+      case message: MemoryWriterCommitMessage => message.data
+      case _ => Seq()


Is there any possible message type here? If not, it's better to throw an exception.

I removed the explicit mapping. I don't want to set the expectation that every data source implementation needs to check and throw exceptions here; it's Spark's responsibility to pass in the right type of message.

SparkQA · 2017-12-13T01:27:31Z

Test build #84803 has finished for PR 19925 at commit 7c46b33.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-12-13T04:04:23Z

Test build #84815 has finished for PR 19925 at commit 8809bf9.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing

Thanks for the updating. Left two comments about unit tests.

zsxwing · 2017-12-13T20:52:31Z

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/RateSourceV2Suite.scala

+    val reader = new RateStreamV2Reader(
+      new DataSourceV2Options(Map("numPartitions" -> "11", "rowsPerSecond" -> "33").asJava))
+    reader.setOffsetRange(Optional.empty(),
+      Optional.of(LongOffset(System.currentTimeMillis() + 1001)))


this will be flaky when the Jenkins is overload. Could you add a clock parameter to RateStreamV2Reader and use ManualClock instead so that we can control the clock?

zsxwing · 2017-12-13T20:54:03Z

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/RateSourceV2Suite.scala

+      case t: RateStreamReadTask =>
+        // Read the first 11 rows. Each partition should be outputting 10 rows per second, so
+        // the 11th row should come 1 second (within a confidence interval) after the first.
+        val startTime = System.currentTimeMillis()


ditto: use ManualClock to avoid flakiness.

SparkQA · 2017-12-13T21:15:03Z

Test build #84874 has finished for PR 19925 at commit 0974ac3.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2017-12-13T23:58:14Z

LGTM pending tests

SparkQA · 2017-12-14T02:11:38Z

Test build #84884 has finished for PR 19925 at commit 3cb6cee.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2017-12-14T06:31:06Z

Thanks! Merging to master to unblock #19926. If there are more comments, we can address them in #19926.

cloud-fan · 2017-12-14T07:33:29Z

sql/core/src/main/java/org/apache/spark/sql/sources/v2/MicroBatchWriteSupport.java

+   * @param queryId A unique string for the writing query. It's possible that there are many writing
+   *                queries running at the same time, and the returned {@link DataSourceV2Writer}
+   *                can use this id to distinguish itself from others.
+   * @param epochId The uniquenumeric ID of the batch within this writing query. This is an


typo: unique numeric

cloud-fan · 2017-12-14T07:38:06Z

sql/core/src/main/java/org/apache/spark/sql/sources/v2/reader/ContinuousDataReader.java

+    /**
+     * Get the offset of the current record, or the start offset if no records have been read.
+     *
+     * The execution engine will call this method along with get() to keep track of the current


better to use a real java doc link, e.g. {@link DataReader#get}

cloud-fan · 2017-12-14T07:40:23Z

sql/core/src/main/java/org/apache/spark/sql/sources/v2/reader/ContinuousReader.java

+import java.util.Optional;
+
+/**
+ * A mix-in interface for {@link DataSourceV2Reader}. Data source readers can implement this


It's not a mix-in interface but a variation on DataSourceV2Reader, right?

cloud-fan · 2017-12-14T07:41:32Z

sql/core/src/main/java/org/apache/spark/sql/sources/v2/reader/MicroBatchReader.java

+import java.util.Optional;
+
+/**
+ * A mix-in interface for {@link DataSourceV2Reader}. Data source readers can implement this


cloud-fan · 2017-12-14T07:42:44Z

sql/core/src/main/java/org/apache/spark/sql/sources/v2/reader/MicroBatchReader.java

+ */
+public interface MicroBatchReader extends DataSourceV2Reader, BaseStreamingSource {
+    /**
+     * Set the desired offset range for read tasks created from this reader. Read tasks will


do you mean this method must be called before createReadTasks?

cloud-fan · 2017-12-14T07:46:37Z

sql/core/src/main/java/org/apache/spark/sql/sources/v2/writer/ContinuousWriter.java

+import org.apache.spark.annotation.InterfaceStability;
+
+/**
+ * A {@link DataSourceV2Writer} for use with continuous stream processing.


A variation of ...?

cloud-fan · 2017-12-14T07:53:26Z

shall we move these new interfaces to org.apache.spark.sql.sources.v2.reader/write.streaming package?

marmbrus · 2017-12-14T16:33:24Z

I would probably do ... streaming.reader/writer if we are going to namespace it.

jose-torres added 24 commits December 7, 2017 14:59

add tests

cbf7837

writer impl

10ac599

rm useless writer

5594f7a

rm weird docs

a364fa6

shuffle around public interfaces

41d732c

fix imports

60c12c2

put deserialize in reader so we don't have to port SerializedOffset

57db413

off by one errors grr

93b6976

document right semantics

abd20db

document checkpoint location

3629d27

add getStart to continuous and clarify semantics

d021f31

cleanup offset set/get docs

052808a

cleanup reader docs

7a7638b

explain getOffset

81efbee

fix fmt

a9c43d9

fix doc

3f03f50

note interfaces are temporary

6338043

fix wording

8949571

lifecycle

df9f224

fix offset semantic implementation

ddaee34

remove unneeded restriction

1608444

deserializeOffset

49525b4

add copyright headers

a9fbf33

rebase against datasource package change

22d07cd

jose-torres force-pushed the continuous-api branch from be0b8e9 to 22d07cd Compare December 7, 2017 23:33

brkyvz reviewed Dec 8, 2017

View reviewed changes

refer properly to sink

0b68873

zsxwing requested changes Dec 13, 2017

View reviewed changes

jose-torres added 2 commits December 12, 2017 16:58

add docs

8a2a4f1

fix problems

8809bf9

even timestamps

0974ac3

zsxwing requested changes Dec 13, 2017

View reviewed changes

jose-torres added 2 commits December 13, 2017 15:28

unify offsets

c1a6322

undo spurious rename

3cb6cee

asfgit closed this in f8c7c1f Dec 14, 2017

cloud-fan reviewed Dec 14, 2017

View reviewed changes

xuanyuanking mentioned this pull request Mar 12, 2019

[SPARK-26594][SQL] DataSourceOptions.asMap should return CaseInsensitiveMap #24062

Closed

ConeyLiu mentioned this pull request Mar 27, 2019

[MINOR] Move java file to java directory #24222

Closed


		package org.apache.spark.sql.sources.v2.reader;

		public abstract class Offset {

[SPARK-22732] Add Structured Streaming APIs to DataSourceV2 #19925

[SPARK-22732] Add Structured Streaming APIs to DataSourceV2 #19925

Uh oh!

Conversation

jose-torres commented Dec 7, 2017

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

SparkQA commented Dec 7, 2017

Uh oh!

SparkQA commented Dec 7, 2017

Uh oh!

SparkQA commented Dec 8, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Dec 11, 2017

Uh oh!

jose-torres commented Dec 11, 2017

Uh oh!

SparkQA commented Dec 12, 2017

Uh oh!

jose-torres commented Dec 12, 2017

Uh oh!

zsxwing left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Dec 13, 2017

Uh oh!

SparkQA commented Dec 13, 2017

Uh oh!

zsxwing left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Dec 13, 2017

Uh oh!

zsxwing commented Dec 13, 2017

Uh oh!

SparkQA commented Dec 14, 2017

Uh oh!

zsxwing commented Dec 14, 2017

Uh oh!

cloud-fan Dec 14, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

cloud-fan Dec 14, 2017 •

edited

Loading