[#940] improvement: Optimize columnar shuffle integration #958

summaryzb · 2023-06-20T09:32:20Z

What changes were proposed in this pull request?

Make it possible to extend uniffle in spark3.
Optimize the shuffleMetric when use columnar shuffle

Why are the changes needed?

#940

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Unit test

summaryzb · 2023-06-20T09:43:00Z

client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java

-    shuffleWriteMetrics.incRecordsWritten(1L);
+    // records is a row based semantic
+    if (isRowBased) {
+      shuffleWriteMetrics.incRecordsWritten(1L);


replace with this

Could we add more comments to explain why we need this?

Is it a common case? I think it binded the implement of gluten. Could we have more common interface?

Comments are added, Yes it's a common case.
All columnar shuffle use it's own serializer, all the serializer related work that is binded with implementation of the columnar framework not limited to gluten should be handled in the rss columnar shuffle writer

Could we move this code to the method addRecord? Columnar shuffle won't call the addRecord method, will it?

That's a good idea

jerqi · 2023-06-20T16:05:56Z

client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java

@@ -141,7 +142,8 @@ public WriteBufferManager(
    this.requireMemoryRetryMax = bufferManagerOptions.getRequireMemoryRetryMax();
    this.arrayOutputStream = new WrappedByteArrayOutputStream(serializerBufferSize);
    // in columnar shuffle, the serializer here is never used
-    if (serializer != null) {
+    this.isRowBased = (serializer != null);


Is it common case? if we support other columnar shuffle like rapids, the other columnar shuffle may use non-null serializer.

Can columnar shuffle be supported by adding configuration or adding a new constructor? This judgment seems a bit hard

In my opinion, all columnar shuffle should pass null serializer to WriterBufferManager.

WriterBufferManager should only do the buffer related work as it's name mean

Columnar data framework is various, the integration work should be done in the implementation of rss columnar shuffle writer

We can pass a non-serializer and ColumnarBatch to addRecord method although we use columnar shuffle. It's not equivalent.

We can pass a non-serializer and ColumnarBatch to addRecord method although we use columnar shuffle. It's not equivalent.

Not only serialization but also partitioning should be handled and are both related with the implementation of the third party columnar framework, we'd better handle them outside of WriterBufferManager

For a shuffle, we can handle partitioning by the partitioner. You mean that partitioner is null for gluten, won't it?
In another word, we may use null serializer row based shuffle.

@jerqi When in columnar shuffle, an element of the iterator is a ColumnarBatch, it's consists of many rows which will be partitioned to different partitionIds, current spark partitioner api cannot handle this scenario.
Actually the java partitioner is fake when use columnar shuffle in gluten, partitioner is implemented in cpp layer

Could we have null serializer for row based shuffle in the future? I think the answer is yes. Because some spark data don't need serialization although the data is organized by row format.

How about indicate row based shuffle false in rssConf @jerqi

jerqi · 2023-06-20T16:06:44Z

client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java

-    shuffleWriteMetrics.incRecordsWritten(1L);
+    // records is a row based semantic
+    if (isRowBased) {
+      shuffleWriteMetrics.incRecordsWritten(1L);


Is it a common case? I think it binded the implement of gluten. Could we have more common interface?

xianjingfeng · 2023-06-21T02:23:59Z

Should we make all fields as protected variables?

LuciferYang · 2023-06-21T03:51:36Z

client-spark/spark3/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java

-  private final Map<Integer, Integer> shuffleIdToNumMapTasks = Maps.newConcurrentMap();
-  private ShuffleManagerGrpcService service;
-  private GrpcServer shuffleManagerServer;
+  protected final String clientType;


why change all fields to protected? If this is necessary, it is also need to add some code comments to explain

Follow the suggestion

summaryzb · 2023-06-24T23:22:13Z

Should we make all fields as protected variables?

No, fix this @xianjingfeng PTAL

codecov-commenter · 2023-06-26T03:05:51Z

Codecov Report

Merging #958 (92a9fc8) into master (8a0ae4b) will decrease coverage by 0.83%.
The diff coverage is 91.66%.

@@             Coverage Diff              @@
##             master     #958      +/-   ##
============================================
- Coverage     55.00%   54.18%   -0.83%     
- Complexity     2466     2474       +8     
============================================
  Files           367      355      -12     
  Lines         19237    17996    -1241     
  Branches       1579     1726     +147     
============================================
- Hits          10582     9751     -831     
+ Misses         8008     7643     -365     
+ Partials        647      602      -45

Impacted Files	Coverage Δ
...pache/spark/shuffle/writer/WriteBufferManager.java	`79.67% <85.71%> (-0.11%)`	⬇️
.../java/org/apache/spark/shuffle/RssSparkConfig.java	`98.18% <100.00%> (+0.05%)`	⬆️

... and 43 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

advancedxy

Another thing about this integration, do we have any integration tests for this, so that we can caught any incompatibility before releasing.

advancedxy · 2023-06-26T03:17:17Z

client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java

@@ -141,7 +142,8 @@ public WriteBufferManager(
    this.requireMemoryRetryMax = bufferManagerOptions.getRequireMemoryRetryMax();
    this.arrayOutputStream = new WrappedByteArrayOutputStream(serializerBufferSize);
    // in columnar shuffle, the serializer here is never used
-    if (serializer != null) {
+    this.isRowBased = (serializer != null);


Can columnar shuffle be supported by adding configuration or adding a new constructor? This judgment seems a bit hard

+1. It seems a bit counterintuitive to rely on serializer == null to check whether the buffer writer manager should support columnar shuffle or not.

Is is possible for gluten to pass additional conf items to spark conf and in Uniffle side we can add a columnarSupport field in the BufferManagerOptions class.

advancedxy · 2023-06-26T03:22:42Z

client-spark/spark3/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java

+  protected AtomicReference<String> id = new AtomicReference<>();
+  protected SparkConf sparkConf;
+  protected ShuffleWriteClient shuffleWriteClient;
+  protected DataPusher dataPusher;


Could you point to the gluten impl again?

It doesn't feel right to just expose these fields here, especially the id and sparkConf fields, they are not exposed in Spark's original shuffle manager.

I think these fields should be accessed at least via a getter/method, so uniffle is free to change its implementation later.

U'r right, id and dataPusher need to be extract to a method.
but sparkConf and shuffleWriteClient are just used in some constructor.

summaryzb · 2023-06-26T05:41:28Z

Another thing about this integration, do we have any integration tests for this, so that we can caught any incompatibility before releasing.

Since Uniffle is kind of dependency for gluten, the integration tests should be placed in gluten side, this will be done after the release. For example we shall not test spark integration in hadoop-common project, acctually we fix the hadoop incompatibility in spark side.
Currently i run the tpcds in our production env to test the integration.

jerqi · 2023-06-26T06:59:58Z

client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java

+    // that is handled by rss shuffle writer implementation
+    if (isRowBased) {
+      shuffleWriteMetrics.incRecordsWritten(1L);
+    }


How about

List<ShuffleBlockInfo> dataPartition = addPartitionData(partitionId, serializedData, serializedDataLength, start); if (isRowBased) { shuffleWriteMetrics.incRecordsWritten(1L); } return dataPartition;

That's better

jerqi · 2023-06-26T07:01:57Z

client-spark/spark3/src/main/java/org/apache/spark/shuffle/writer/RssShuffleWriter.java

@@ -127,7 +127,7 @@ public RssShuffleWriter(
      ShuffleWriteClient shuffleWriteClient,
      RssShuffleHandle<K, V, C> rssHandle,
      Function<String, Boolean> taskFailureCallback) {
-    LOG.warn("RssShuffle start write taskAttemptId data" + taskAttemptId);
+    LOG.warn("RssShuffleaskAttempt start write taskAttemptId data" + taskAttemptId);


There is a misspell.

jerqi

LGTM, let @loukey-lj take another look.

summaryzb · 2023-06-28T04:51:12Z

gentle ping @xianjingfeng PTAL

xianjingfeng

LGTM

xianjingfeng · 2023-06-28T06:06:37Z

Merged. Thanks all.

make columnar shuffle extend uniffle

39bf629

summaryzb changed the title ~~[#940] improvement: make the shuffle~~ [#940] improvement: Optimize columnar shuffle integration Jun 20, 2023

jerqi requested review from LuciferYang and xianjingfeng June 20, 2023 09:38

summaryzb commented Jun 20, 2023

View reviewed changes

jerqi reviewed Jun 20, 2023

View reviewed changes

LuciferYang reviewed Jun 21, 2023

View reviewed changes

add comments

5326c92

advancedxy reviewed Jun 26, 2023

View reviewed changes

make some code change

8ded651

jerqi reviewed Jun 26, 2023

View reviewed changes

summaryzb added 2 commits June 26, 2023 15:51

fix misspell

a1f479c

add rss config option

f20d6b5

jerqi requested a review from loukey-lj June 27, 2023 06:55

fix test

92a9fc8

jerqi approved these changes Jun 27, 2023

View reviewed changes

loukey-lj approved these changes Jun 27, 2023

View reviewed changes

xianjingfeng approved these changes Jun 28, 2023

View reviewed changes

xianjingfeng merged commit 0a42cfb into apache:master Jun 28, 2023
27 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[#940] improvement: Optimize columnar shuffle integration #958

[#940] improvement: Optimize columnar shuffle integration #958

summaryzb commented Jun 20, 2023

summaryzb Jun 20, 2023

jerqi Jun 20, 2023

jerqi Jun 20, 2023

summaryzb Jun 24, 2023

jerqi Jun 25, 2023

summaryzb Jun 26, 2023

jerqi Jun 20, 2023

loukey-lj Jun 21, 2023

LuciferYang Jun 21, 2023

summaryzb Jun 24, 2023

jerqi Jun 25, 2023 •

edited

Loading

jerqi Jun 26, 2023

summaryzb Jun 26, 2023

jerqi Jun 26, 2023 •

edited

Loading

summaryzb Jun 27, 2023

jerqi Jun 27, 2023

jerqi Jun 20, 2023

xianjingfeng commented Jun 21, 2023

LuciferYang Jun 21, 2023 •

edited

Loading

summaryzb Jun 24, 2023

summaryzb commented Jun 24, 2023 •

edited

Loading

codecov-commenter commented Jun 26, 2023 •

edited

Loading

advancedxy left a comment

advancedxy Jun 26, 2023

advancedxy Jun 26, 2023

summaryzb Jun 26, 2023

summaryzb commented Jun 26, 2023

jerqi Jun 26, 2023 •

edited

Loading

summaryzb Jun 26, 2023

jerqi Jun 26, 2023

jerqi left a comment

summaryzb commented Jun 28, 2023

xianjingfeng left a comment

xianjingfeng commented Jun 28, 2023

[#940] improvement: Optimize columnar shuffle integration #958

[#940] improvement: Optimize columnar shuffle integration #958

Conversation

summaryzb commented Jun 20, 2023

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jerqi Jun 25, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jerqi Jun 26, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xianjingfeng commented Jun 21, 2023

LuciferYang Jun 21, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

summaryzb commented Jun 24, 2023 • edited Loading

codecov-commenter commented Jun 26, 2023 • edited Loading

Codecov Report

advancedxy left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

summaryzb commented Jun 26, 2023

jerqi Jun 26, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jerqi left a comment

Choose a reason for hiding this comment

summaryzb commented Jun 28, 2023

xianjingfeng left a comment

Choose a reason for hiding this comment

xianjingfeng commented Jun 28, 2023

jerqi Jun 25, 2023 •

edited

Loading

jerqi Jun 26, 2023 •

edited

Loading

LuciferYang Jun 21, 2023 •

edited

Loading

summaryzb commented Jun 24, 2023 •

edited

Loading

codecov-commenter commented Jun 26, 2023 •

edited

Loading

jerqi Jun 26, 2023 •

edited

Loading