[#2086] feat(spark): Support cut partition to slices and served by multiply server

[maobaolong]

What changes were proposed in this pull request?

Support sliced store partition to multiply server.

Limitation:

• Only finished tested the netty mode.

Why are the changes needed?

Fix: #2086

Does this PR introduce any user-facing change?

No.

How was this patch tested?

• Start multiply servers and coordinator on local
• Start a spark standalone env on local
• Start spark-shell and execute test.scala

bin/spark-shell  --master  spark://localhost:7077  --deploy-mode client --conf spark.rss.client.reassign.blockRetryMaxTimes=3 --conf spark.rss.writer.buffer.spill.size=30 --conf spark.rss.client.reassign.enabled=true  --conf spark.shuffle.manager=org.apache.spark.shuffle.RssShuffleManager --conf spark.rss.coordinator.quorum=localhost:19999  --conf spark.rss.storage.type=LOCALFILE --conf spark.serializer=org.apache.spark.serializer.KryoSerializer --conf spark.rss.test.mode.enable=true --conf spark.rss.client.type=GRPC_NETTY --conf spark.sql.shuffle.partitions=1  -i test.scala

• test.scala

val data = sc.parallelize(Seq(("A", 1), ("B", 2), ("C", 3), ("A", 4), ("B", 5), ("A", 6), ("A", 7),("A", 7), ("A", 7), ("A", 7), ("A", 7), ("A", 7), ("A", 7), ("A", 7), ("A", 7), ("A", 7), ("A", 7), ("A", 7), ("A", 7)));
val result = data.reduceByKey(_ + _);
result.collect().foreach(println);
System.exit(0);

[github-actions]

Test Results

 2 926 files  ± 0   2 926 suites  ±0   6h 11m 57s ⏱️ - 13m 10s
 1 090 tests + 2   1 088 ✅ + 2   2 💤 ±0  0 ❌ ±0 
13 655 runs  +25  13 625 ✅ +25  30 💤 ±0  0 ❌ ±0 

Results for commit 949d57f. ± Comparison against base commit 4206541.

♻️ This comment has been updated with latest results.

[zuston]

Great work. Thanks for your effort. Left some comments.

[zuston]

If using list instead of set, is it possible to store the duplicate replacement servers?

[maobaolong]

Before this change, what the set of excludedServerToReplacements stored the servers about?

[maobaolong]

Revert the set -> list change.

[zuston]

why not marking the partitionIds name as the needSplitPartitionIds

[maobaolong]

Done.

[zuston]

Why not extract the general logic to limit the partitioned data? I think the partition split and huge partition limit could be scoped in the same abstract limit policies.

[maobaolong]

Address this after other comments addressed and tested.

[maobaolong]

@zuston I found there are something wrong with excludedServerToReplacements for split partition, as the server is not faulty server only just cannot store more data for specific partition, so it may should not be added to the exclude servers.

Reference to

  public Set<String> listExcludedServers() {
    return excludedServerToReplacements.keySet();
  }

[zuston]

Overall lgtm. I hope some test cases could be added to cover this case.

[zuston]

Overall lgtm. I hope some test cases could be added to cover this case.

Have any update on this? @maobaolong If you have applied in your internal cluster, please let me know. Thanks

[maobaolong]

@zuston Yeah, I test this on our pressure test cluster and verified by our pressure test case. We have 10 node in that cluster, and run a 17TiB shuffle data spark application with 3 shuffle stage, there are 2000 partition for each shuffle stage. I verified the shuffle split result by #2133 and #2156.

BTW, #2137 this PR make me easy to dynamic update conf and happy on test this feature.

[maobaolong]

@zuston UT has been added, would you like to take another look?

[zuston]

LGTM.

[maobaolong]

@zuston Thank you so much for the discussion and reviewing for this feature, I cannot finish this work without your help!

Merge it now!