[WIP][feature][spark] Support streaming

[CheneyYin]

Purpose of this pull request

Support streaming for spark engine.
Related:

• [Spark streaming] fix spark streaming #4713
• [[Feature]]Spark Support cdc #5750

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config
• Update the release-note.

[hailin0]

cc @Carl-Zhou-CN

[CheneyYin]

seatunnel/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-3.3/src/main/java/org/apache/seatunnel/translation/spark/source/partition/micro/SeaTunnelMicroBatchPartitionReader.java

Lines 27 to 49 in 1bba723

	public class SeaTunnelMicroBatchPartitionReader implements PartitionReader<InternalRow> {
	private final ParallelBatchPartitionReader partitionReader;
	public SeaTunnelMicroBatchPartitionReader(ParallelBatchPartitionReader partitionReader) {
	this.partitionReader = partitionReader;
	}
	@Override
	public boolean next() throws IOException {
	return partitionReader.next();
	}
	@Override
	public InternalRow get() {
	return partitionReader.get();
	}
	@Override
	public void close() throws IOException {
	partitionReader.close();
	}
	}

seatunnel/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-3.3/src/main/java/org/apache/seatunnel/translation/spark/source/partition/batch/ParallelBatchPartitionReader.java

Lines 87 to 97 in 1bba723

	public boolean next() throws IOException {
	prepare();
	while (running && handover.isEmpty()) {
	try {
	Thread.sleep(INTERVAL);
	} catch (InterruptedException e) {
	throw new RuntimeException(e);
	}
	}
	return running || !handover.isEmpty();
	}

PartitionReader never close in streaming mode.

[Carl-Zhou-CN]

seatunnel/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-3.3/src/main/java/org/apache/seatunnel/translation/spark/source/partition/micro/SeaTunnelMicroBatchPartitionReader.java

Lines 27 to 49 in 1bba723

	public class SeaTunnelMicroBatchPartitionReader implements PartitionReader<InternalRow> {
	private final ParallelBatchPartitionReader partitionReader;
	public SeaTunnelMicroBatchPartitionReader(ParallelBatchPartitionReader partitionReader) {
	this.partitionReader = partitionReader;
	}
	@Override
	public boolean next() throws IOException {
	return partitionReader.next();
	}
	@Override
	public InternalRow get() {
	return partitionReader.get();
	}
	@Override
	public void close() throws IOException {
	partitionReader.close();
	}
	}

seatunnel/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-3.3/src/main/java/org/apache/seatunnel/translation/spark/source/partition/batch/ParallelBatchPartitionReader.java

Lines 87 to 97 in 1bba723

	public boolean next() throws IOException {
	prepare();
	while (running && handover.isEmpty()) {
	try {
	Thread.sleep(INTERVAL);
	} catch (InterruptedException e) {
	throw new RuntimeException(e);
	}
	}
	return running || !handover.isEmpty();
	}

PartitionReader never close in streaming mode.

hi @CheneyYin It seems that after a checkpoint, it will be close

[CheneyYin]

seatunnel/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-3.3/src/main/java/org/apache/seatunnel/translation/spark/source/partition/micro/SeaTunnelMicroBatchPartitionReader.java

Lines 27 to 49 in 1bba723

	public class SeaTunnelMicroBatchPartitionReader implements PartitionReader<InternalRow> {
	private final ParallelBatchPartitionReader partitionReader;
	public SeaTunnelMicroBatchPartitionReader(ParallelBatchPartitionReader partitionReader) {
	this.partitionReader = partitionReader;
	}
	@Override
	public boolean next() throws IOException {
	return partitionReader.next();
	}
	@Override
	public InternalRow get() {
	return partitionReader.get();
	}
	@Override
	public void close() throws IOException {
	partitionReader.close();
	}
	}

seatunnel/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-3.3/src/main/java/org/apache/seatunnel/translation/spark/source/partition/batch/ParallelBatchPartitionReader.java

Lines 87 to 97 in 1bba723

	public boolean next() throws IOException {
	prepare();
	while (running && handover.isEmpty()) {
	try {
	Thread.sleep(INTERVAL);
	} catch (InterruptedException e) {
	throw new RuntimeException(e);
	}
	}
	return running || !handover.isEmpty();
	}

PartitionReader never close in streaming mode.

hi @CheneyYin It seems that after a checkpoint, it will be close

Yes. If the reader does not receive new data for a long time, Spark will end the current micro batch. Spark's micro batch mechanism does not fully meet the requirements of long term streaming computing. First, creating a new reader for the next batch will incur some overhead. Second, the granularity of fault recovery is too large, and the Spark micro batch mechanism cannot restore the reader from the latest snapshot of the Seatunnel reader.
I am looking for strategies to alleviate these problems while ensuring fault recovery. Currently, I add metadata to the seatunnel row and use a special identifier to represent the checkpoint event. After the source completes a checkpoint, it will create a checkpoint record and send it to the downstream. After receiving the checkpoint record, the sink saves the snapshot and confirms the prepared checkpoint made by the source. These checkpoint operations are performed based on the file system directory space.

[CheneyYin]

The checkpoint space like this:

./
├── commits
│   ├........
│   ├── 10
│   ├── 11
│   ├── 12
│   ├── 13
│   ├── 14
│   ├── 15
|    ......
├── metadata
├── offsets
│   ├ ........
│   ├── 10
│   ├── 11
│   ├── 12
│   ├── 13
│   ├── 14
│   ├── 15
│   ├── 16
│     .......
└── sources
    └── 0
        ├ ........
        ├── 11
        │   ├── 0
        │   │   └── 0.committed
        │   └── 1
        │       └── 0.committed
        ├── 12
        │   ├── 0
        │   │   └── 0.committed
        │   └── 1
        │       └── 0.committed
        ├── 13
        │   ├── 0
        │   │   └── 0.committed
        │   └── 1
        │       └── 0.committed
        ├── 14
        │   ├── 0
        │   │   └── 0.committed
        │   └── 1
        │       └── 0.committed
        ├── 15
        │   ├── 0
        │   │   └── 0.committed
        │   └── 1
        │       └── 0.committed
  ........

[CheneyYin]

seatunnel/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-3.3/src/main/java/org/apache/seatunnel/translation/spark/source/partition/micro/SeaTunnelMicroBatchPartitionReader.java

Lines 27 to 49 in 1bba723

	public class SeaTunnelMicroBatchPartitionReader implements PartitionReader<InternalRow> {
	private final ParallelBatchPartitionReader partitionReader;
	public SeaTunnelMicroBatchPartitionReader(ParallelBatchPartitionReader partitionReader) {
	this.partitionReader = partitionReader;
	}
	@Override
	public boolean next() throws IOException {
	return partitionReader.next();
	}
	@Override
	public InternalRow get() {
	return partitionReader.get();
	}
	@Override
	public void close() throws IOException {
	partitionReader.close();
	}
	}

seatunnel/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-3.3/src/main/java/org/apache/seatunnel/translation/spark/source/partition/batch/ParallelBatchPartitionReader.java

Lines 87 to 97 in 1bba723

	public boolean next() throws IOException {
	prepare();
	while (running && handover.isEmpty()) {
	try {
	Thread.sleep(INTERVAL);
	} catch (InterruptedException e) {
	throw new RuntimeException(e);
	}
	}
	return running || !handover.isEmpty();
	}

PartitionReader never close in streaming mode.

hi @CheneyYin It seems that after a checkpoint, it will be close

The next() never return false useless call close(). However Spark MicroStreamExecution will call close() after next() return false. So reader never stop and never commit batch.

[Carl-Zhou-CN]

seatunnel/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-3.3/src/main/java/org/apache/seatunnel/translation/spark/source/partition/micro/SeaTunnelMicroBatchPartitionReader.java

Lines 27 to 49 in 1bba723

	public class SeaTunnelMicroBatchPartitionReader implements PartitionReader<InternalRow> {
	private final ParallelBatchPartitionReader partitionReader;
	public SeaTunnelMicroBatchPartitionReader(ParallelBatchPartitionReader partitionReader) {
	this.partitionReader = partitionReader;
	}
	@Override
	public boolean next() throws IOException {
	return partitionReader.next();
	}
	@Override
	public InternalRow get() {
	return partitionReader.get();
	}
	@Override
	public void close() throws IOException {
	partitionReader.close();
	}
	}

seatunnel/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-3.3/src/main/java/org/apache/seatunnel/translation/spark/source/partition/batch/ParallelBatchPartitionReader.java

Lines 87 to 97 in 1bba723

	public boolean next() throws IOException {
	prepare();
	while (running && handover.isEmpty()) {
	try {
	Thread.sleep(INTERVAL);
	} catch (InterruptedException e) {
	throw new RuntimeException(e);
	}
	}
	return running || !handover.isEmpty();
	}

PartitionReader never close in streaming mode.

hi @CheneyYin It seems that after a checkpoint, it will be close

Yes. If the reader does not receive new data for a long time, Spark will end the current micro batch. Spark's micro batch mechanism does not fully meet the requirements of long term streaming computing. First, creating a new reader for the next batch will incur some overhead. Second, the granularity of fault recovery is too large, and the Spark micro batch mechanism cannot restore the reader from the latest snapshot of the Seatunnel reader. I am looking for strategies to alleviate these problems while ensuring fault recovery. Currently, I add metadata to the seatunnel row and use a special identifier to represent the checkpoint event. After the source completes a checkpoint, it will create a checkpoint record and send it to the downstream. After receiving the checkpoint record, the sink saves the snapshot and confirms the prepared checkpoint made by the source. These checkpoint operations are performed based on the file system directory space.

Yes, the micro-batch process cannot meet the requirements of streaming.

[Carl-Zhou-CN]

seatunnel/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-3.3/src/main/java/org/apache/seatunnel/translation/spark/source/partition/micro/SeaTunnelMicroBatchPartitionReader.java

Lines 27 to 49 in 1bba723

	public class SeaTunnelMicroBatchPartitionReader implements PartitionReader<InternalRow> {
	private final ParallelBatchPartitionReader partitionReader;
	public SeaTunnelMicroBatchPartitionReader(ParallelBatchPartitionReader partitionReader) {
	this.partitionReader = partitionReader;
	}
	@Override
	public boolean next() throws IOException {
	return partitionReader.next();
	}
	@Override
	public InternalRow get() {
	return partitionReader.get();
	}
	@Override
	public void close() throws IOException {
	partitionReader.close();
	}
	}

seatunnel/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-3.3/src/main/java/org/apache/seatunnel/translation/spark/source/partition/batch/ParallelBatchPartitionReader.java

Lines 87 to 97 in 1bba723

	public boolean next() throws IOException {
	prepare();
	while (running && handover.isEmpty()) {
	try {
	Thread.sleep(INTERVAL);
	} catch (InterruptedException e) {
	throw new RuntimeException(e);
	}
	}
	return running || !handover.isEmpty();
	}

PartitionReader never close in streaming mode.

hi @CheneyYin It seems that after a checkpoint, it will be close

The next() never return false useless call close(). However Spark MicroStreamExecution will call close() after next() return false. So reader never stop and never commit batch.

yes, you're right

[Carl-Zhou-CN]

seatunnel/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-3.3/src/main/java/org/apache/seatunnel/translation/spark/source/partition/micro/SeaTunnelMicroBatchPartitionReader.java

Lines 27 to 49 in 1bba723

	public class SeaTunnelMicroBatchPartitionReader implements PartitionReader<InternalRow> {
	private final ParallelBatchPartitionReader partitionReader;
	public SeaTunnelMicroBatchPartitionReader(ParallelBatchPartitionReader partitionReader) {
	this.partitionReader = partitionReader;
	}
	@Override
	public boolean next() throws IOException {
	return partitionReader.next();
	}
	@Override
	public InternalRow get() {
	return partitionReader.get();
	}
	@Override
	public void close() throws IOException {
	partitionReader.close();
	}
	}

seatunnel/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-3.3/src/main/java/org/apache/seatunnel/translation/spark/source/partition/batch/ParallelBatchPartitionReader.java

Lines 87 to 97 in 1bba723

	public boolean next() throws IOException {
	prepare();
	while (running && handover.isEmpty()) {
	try {
	Thread.sleep(INTERVAL);
	} catch (InterruptedException e) {
	throw new RuntimeException(e);
	}
	}
	return running || !handover.isEmpty();
	}

PartitionReader never close in streaming mode.

hi @CheneyYin It seems that after a checkpoint, it will be close

Yes. If the reader does not receive new data for a long time, Spark will end the current micro batch. Spark's micro batch mechanism does not fully meet the requirements of long term streaming computing. First, creating a new reader for the next batch will incur some overhead. Second, the granularity of fault recovery is too large, and the Spark micro batch mechanism cannot restore the reader from the latest snapshot of the Seatunnel reader. I am looking for strategies to alleviate these problems while ensuring fault recovery. Currently, I add metadata to the seatunnel row and use a special identifier to represent the checkpoint event. After the source completes a checkpoint, it will create a checkpoint record and send it to the downstream. After receiving the checkpoint record, the sink saves the snapshot and confirms the prepared checkpoint made by the source. These checkpoint operations are performed based on the file system directory space.

I think this pattern would be more like Spark's continuous streaming mode, but it seems to completely lack fault tolerance

[CheneyYin]

seatunnel/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-3.3/src/main/java/org/apache/seatunnel/translation/spark/source/partition/micro/SeaTunnelMicroBatchPartitionReader.java

Lines 27 to 49 in 1bba723

	public class SeaTunnelMicroBatchPartitionReader implements PartitionReader<InternalRow> {
	private final ParallelBatchPartitionReader partitionReader;
	public SeaTunnelMicroBatchPartitionReader(ParallelBatchPartitionReader partitionReader) {
	this.partitionReader = partitionReader;
	}
	@Override
	public boolean next() throws IOException {
	return partitionReader.next();
	}
	@Override
	public InternalRow get() {
	return partitionReader.get();
	}
	@Override
	public void close() throws IOException {
	partitionReader.close();
	}
	}

seatunnel/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-3.3/src/main/java/org/apache/seatunnel/translation/spark/source/partition/batch/ParallelBatchPartitionReader.java

Lines 87 to 97 in 1bba723

	public boolean next() throws IOException {
	prepare();
	while (running && handover.isEmpty()) {
	try {
	Thread.sleep(INTERVAL);
	} catch (InterruptedException e) {
	throw new RuntimeException(e);
	}
	}
	return running || !handover.isEmpty();
	}

PartitionReader never close in streaming mode.

hi @CheneyYin It seems that after a checkpoint, it will be close

Yes. If the reader does not receive new data for a long time, Spark will end the current micro batch. Spark's micro batch mechanism does not fully meet the requirements of long term streaming computing. First, creating a new reader for the next batch will incur some overhead. Second, the granularity of fault recovery is too large, and the Spark micro batch mechanism cannot restore the reader from the latest snapshot of the Seatunnel reader. I am looking for strategies to alleviate these problems while ensuring fault recovery. Currently, I add metadata to the seatunnel row and use a special identifier to represent the checkpoint event. After the source completes a checkpoint, it will create a checkpoint record and send it to the downstream. After receiving the checkpoint record, the sink saves the snapshot and confirms the prepared checkpoint made by the source. These checkpoint operations are performed based on the file system directory space.

I think this pattern would be more like Spark's continuous streaming mode, but it seems to completely lack fault tolerance

It can ensure end-to-end at-least-once semantics. If sink can be idempotent for handling reprocessing data, it can ensure exactly-once. At present, spark continuous streaming execution mode is still experimental and guarantees at-least-once fault-tolerance.

[hailin0]

❤️