Skip to content
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

Jump to bottom

[Feature] make data block balance before importing data #5026 #5044

Closed
wants to merge 7 commits into from
Closed

[Feature] make data block balance before importing data #5026 #5044

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
e951847
make data block balance before importing data #5026
Jul 7, 2023
8693bf1
doc for partition_balance #5026
Jul 10, 2023
35bb0de
doc for partition_balance #5026
Jul 11, 2023
a6b1069
doc for partition_balance
Jul 11, 2023
bdab8d6
Change the way for getting configuration
Jul 19, 2023
112d1ce
Change the way for getting configuration
Jul 20, 2023
29485a9
fix coding error
Aug 7, 2023
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
19 changes: 15 additions & 4 deletions docs/en/connector-v2/sink/common-options.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,11 @@

> Common parameters of sink connectors

| name | type | required | default value |
|-------------------|--------|----------|---------------|
| source_table_name | string | no | - |
| parallelism | int | no | - |
| name | type | required | default value |
|-------------------|---------|----------|---------------|
| source_table_name | string | no | - |
| parallelism | int | no | - |
| partition_balance | boolean | no | false |

### source_table_name [string]

Expand All @@ -19,6 +20,16 @@ When `parallelism` is not specified, the `parallelism` in env is used by default

When parallelism is specified, it will override the parallelism in env.

### partition_balance [boolean]
Copy link
Member

Choose a reason for hiding this comment

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

Please add comment that this option only supported by Flink/Spark.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

done


When `partition_balance` is set to true, in the sink process, a repartition will be performed first to ensure that the size of each partition is roughly the same, which can avoid problems caused by data skew, but it will consume some extra time.

The default value is false, support Spark and Flink engine

When `partition_balance` is not specified, the `partition_balance` in env is used by default.

When `partition_balance` is specified, it will override the `partition_balance` in env.

## Examples

```bash
Expand Down
9 changes: 9 additions & 0 deletions seatunnel-api/src/main/java/org/apache/seatunnel/api/common/CommonOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,4 +66,13 @@ public interface CommonOptions {
.withDescription(
"When parallelism is not specified, the parallelism in env is used by default. "
+ "When parallelism is specified, it will override the parallelism in env.");

Option<Boolean> PARTITION_BALANCE =
Options.key("partition_balance")
.booleanType()
.defaultValue(false)
.withDescription(
"When partition_balance is set to true, "
+ "in the sink process, a repartition will be performed first to ensure that the size of each partition is roughly the same, "
+ "which can avoid problems caused by data skew, but it will consume some extra time. The default value is false");
}
6 changes: 6 additions & 0 deletions ...src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,12 @@ public List<DataStream<Row>> execute(List<DataStream<Row>> upstreamDataStreams)
DataSaveMode dataSaveMode = saveModeSink.getDataSaveMode();
saveModeSink.handleSaveMode(dataSaveMode);
}
if (sinkConfig.hasPath(CommonOptions.PARTITION_BALANCE.key())) {
Boolean needBalance = sinkConfig.getBoolean(CommonOptions.PARTITION_BALANCE.key());
if (needBalance) {
stream = stream.shuffle();
}
}
DataStreamSink<Row> dataStreamSink =
stream.sinkTo(SinkV1Adapter.wrap(new FlinkSink<>(seaTunnelSink)))
.name(seaTunnelSink.getPluginName());
Expand Down
6 changes: 6 additions & 0 deletions ...src/main/java/org/apache/seatunnel/core/starter/spark/execution/SinkExecuteProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,12 @@ public List<Dataset<Row>> execute(List<Dataset<Row>> upstreamDataStreams)
CommonOptions.PARALLELISM.key(),
CommonOptions.PARALLELISM.defaultValue());
}
if (sinkConfig.hasPath(CommonOptions.PARTITION_BALANCE.key())) {
boolean needBalance = sinkConfig.getBoolean(CommonOptions.PARTITION_BALANCE.key());
if (needBalance) {
dataset = dataset.repartition(parallelism);
}
}
dataset.sparkSession().read().option(CommonOptions.PARALLELISM.key(), parallelism);
// TODO modify checkpoint location
seaTunnelSink.setTypeInfo(
Expand Down
Loading