Skip to content

[#2362] feat(client): Add support of zstd parallel compression #2363

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zuston merged 2 commits into from
Feb 8, 2025
Merged

[#2362] feat(client): Add support of zstd parallel compression #2363

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f812cfa
feat(#2362): Add support of zstd parallel compression
Feb 7, 2025
fce4cc5
add doc
Feb 7, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions common/src/main/java/org/apache/uniffle/common/compression/Codec.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
import org.apache.uniffle.common.config.RssConf;

import static org.apache.uniffle.common.config.RssClientConf.COMPRESSION_TYPE;
import static org.apache.uniffle.common.config.RssClientConf.ZSTD_COMPRESSION_LEVEL;

public abstract class Codec {

Expand All @@ -33,7 +32,7 @@ public static Optional<Codec> newInstance(RssConf rssConf) {
case NONE:
return Optional.empty();
case ZSTD:
return Optional.of(ZstdCodec.getInstance(rssConf.get(ZSTD_COMPRESSION_LEVEL)));
return Optional.of(ZstdCodec.getInstance(rssConf));
case SNAPPY:
return Optional.of(SnappyCodec.getInstance());
case NOOP:
Expand Down
22 changes: 19 additions & 3 deletions common/src/main/java/org/apache/uniffle/common/compression/ZstdCodec.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,22 +20,29 @@
import java.nio.ByteBuffer;

import com.github.luben.zstd.Zstd;
import com.github.luben.zstd.ZstdCompressCtx;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.uniffle.common.config.RssConf;
import org.apache.uniffle.common.exception.RssException;

import static org.apache.uniffle.common.config.RssClientConf.ZSTD_COMPRESSION_LEVEL;
import static org.apache.uniffle.common.config.RssClientConf.ZSTD_COMPRESSION_WORKER_NUMBER;

public class ZstdCodec extends Codec {
private static final Logger LOGGER = LoggerFactory.getLogger(ZstdCodec.class);

private int compressionLevel;
private int workerNumber;

private static class LazyHolder {
static final ZstdCodec INSTANCE = new ZstdCodec();
}

public static ZstdCodec getInstance(int level) {
LazyHolder.INSTANCE.compressionLevel = level;
public static ZstdCodec getInstance(RssConf conf) {
LazyHolder.INSTANCE.compressionLevel = conf.get(ZSTD_COMPRESSION_LEVEL);
LazyHolder.INSTANCE.workerNumber = conf.get(ZSTD_COMPRESSION_WORKER_NUMBER);
return LazyHolder.INSTANCE;
}

Expand Down Expand Up @@ -69,7 +76,16 @@ public void decompress(ByteBuffer src, int uncompressedLen, ByteBuffer dst, int

@Override
public byte[] compress(byte[] src) {
return Zstd.compress(src, compressionLevel);
ZstdCompressCtx ctx = new ZstdCompressCtx();
Copy link
Contributor

@roryqi roryqi Feb 7, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One question: Do we need to create Ctx for every time? It will cost time if ZstdCompressCtx will create and destroy the thread.

Copy link
Member Author

@zuston zuston Feb 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice catch. Actually the code of Zstd.compress(src, compressionLevel); also will create ctx everytime. Let me dig this deeply

Copy link
Contributor

@roryqi roryqi Feb 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's ok if the default context don't need to create extra threads actually. Just feel that we should consider more about this place.

Copy link
Member Author

@zuston zuston Feb 7, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, the compression ctx could be reused.

try {
ctx.setLevel(compressionLevel);
if (workerNumber > 0) {
ctx.setWorkers(workerNumber);
}
return ctx.compress(src);
} finally {
ctx.close();
}
}

@Override
Expand Down
7 changes: 7 additions & 0 deletions common/src/main/java/org/apache/uniffle/common/config/RssClientConf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,13 @@ public class RssClientConf {
.defaultValue(3)
.withDescription("The zstd compression level, the default level is 3");

public static final ConfigOption<Integer> ZSTD_COMPRESSION_WORKER_NUMBER =
ConfigOptions.key("rss.client.io.compression.zstd.workerNumber")
.intType()
.defaultValue(-1)
.withDescription(
"Set the parallel compression worker number. This will not enabled by default");

public static final ConfigOption<ShuffleDataDistributionType> DATA_DISTRIBUTION_TYPE =
ConfigOptions.key("rss.client.shuffle.data.distribution.type")
.enumType(ShuffleDataDistributionType.class)
Expand Down
1 change: 1 addition & 0 deletions docs/client_guide/client_guide.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ The important configuration of client is listed as following. These configuratio
| <client_type>.rss.client.assignment.shuffle.nodes.max | -1 | The number of required assignment shuffle servers. If it is less than 0 or equals to 0 or greater than the coordinator's config of "rss.coordinator.shuffle.nodes.max", it will use the size of "rss.coordinator.shuffle.nodes.max" default |
| <client_type>.rss.client.io.compression.codec | lz4 | The compression codec is used to compress the shuffle data. Default codec is `lz4`. Other options are`ZSTD` and `SNAPPY`. |
| <client_type>.rss.client.io.compression.zstd.level | 3 | The zstd compression level, the default level is 3 |
| <client_type>.rss.client.io.compression.zstd.workerNumber | -1 | Set zstd parallel compression worker number. This will not enabled by default |
| <client_type>.rss.client.shuffle.data.distribution.type | NORMAL | The type of partition shuffle data distribution, including normal and local_order. The default value is normal. Now this config is only valid in Spark3.x |
| <client_type>.rss.estimate.task.concurrency.dynamic.factor | 1.0 | Between 0 and 1, used to estimate task concurrency, when the client is spark, it represents how likely is this part of the resource between spark.dynamicAllocation.minExecutors and spark.dynamicAllocation.maxExecutors to be allocated, when the client is mr, it represents how likely the resources of map and reduce are satisfied. Effective when <client_type>.rss.estimate.server.assignment.enabled=true or Coordinator's rss.coordinator.select.partition.strategy is CONTINUOUS. |
| <client_type>.rss.estimate.server.assignment.enabled | false | Support mr and spark, whether to enable estimation of the number of ShuffleServers that need to be allocated based on the number of concurrent tasks. |
Expand Down
Loading