Skip to content

Add a parameter that controls the number of StreamLoad tasks committed per partition #92 #99

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

Open
baishaoisde wants to merge 10 commits into
base: master
Choose a base branch
from
Open

Add a parameter that controls the number of StreamLoad tasks committed per partition #92 #99

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
3a1aed9
新增分区任务原子性控制参数，参数默认值为false,若为false,则为每个分区可执行多次StreamLoad,若为true,则限制每个分…
baishaoisde Apr 6, 2023
9b0efff
fix
baishaoisde May 17, 2023
8e87ed2
Merge branch 'master' into baishaode_pr
baishaoisde May 29, 2023
4538a43
[refactor] Unified writing through DorisWriter (#104)
gnehil May 30, 2023
84dbb2c
Update version (#106)
gnehil Jun 5, 2023
9291a2e
Update pom.xml (#107)
caoliang-web Jun 5, 2023
53525c2
Pom update (#108)
JNSimba Jun 5, 2023
83fdd53
Added atomicity control parameters for partition task, Parameter defa…
baishaoisde Apr 6, 2023
734ef75
Merge remote-tracking branch 'origin/baishaode_pr' into baishaode_pr
baishaoisde Jun 6, 2023
6e1f6bb
resolve conflict
baishaoisde Jun 6, 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
6 changes: 6 additions & 0 deletions spark-doris-connector/src/main/java/org/apache/doris/spark/cfg/ConfigurationOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,4 +85,10 @@ public interface ConfigurationOptions {
String DORIS_SINK_TASK_USE_REPARTITION = "doris.sink.task.use.repartition";

boolean DORIS_SINK_TASK_USE_REPARTITION_DEFAULT = false;

//设置每个分区仅提交一个StreamLoad任务，以保证任务失败时task重试不会导致对同一批数据重复提交StreamLoad任务。
Copy link
Member

@JNSimba JNSimba May 26, 2023

Choose a reason for hiding this comment

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

english comment

Copy link
Author

@baishaoisde baishaoisde May 29, 2023

Choose a reason for hiding this comment

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

Ok, I will change it to English.

String DORIS_SINK_PER_PARTITION_TASK_ATOMICITY = "doris.sink.per.partition.task.atomicity";

boolean DORIS_SINK_PER_PARTITION_TASK_ATOMICITY_DEFAULT = false;

}
4 changes: 3 additions & 1 deletion spark-doris-connector/src/main/scala/org/apache/doris/spark/sql/DorisSourceProvider.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@ private[sql] class DorisSourceProvider extends DataSourceRegister
val maxRetryTimes = sparkSettings.getIntegerProperty(ConfigurationOptions.DORIS_SINK_MAX_RETRIES, ConfigurationOptions.SINK_MAX_RETRIES_DEFAULT)
val sinkTaskPartitionSize = sparkSettings.getIntegerProperty(ConfigurationOptions.DORIS_SINK_TASK_PARTITION_SIZE)
val sinkTaskUseRepartition = sparkSettings.getProperty(ConfigurationOptions.DORIS_SINK_TASK_USE_REPARTITION, ConfigurationOptions.DORIS_SINK_TASK_USE_REPARTITION_DEFAULT.toString).toBoolean
val partitionTaskAtomicity = settings.getProperty(ConfigurationOptions.DORIS_SINK_PER_PARTITION_TASK_ATOMICITY, ConfigurationOptions.DORIS_SINK_PER_PARTITION_TASK_ATOMICITY_DEFAULT.toString).toBoolean

logger.info(s"maxRowCount ${maxRowCount}")
logger.info(s"maxRetryTimes ${maxRetryTimes}")
Expand All @@ -83,10 +84,11 @@ private[sql] class DorisSourceProvider extends DataSourceRegister
line.add(field.asInstanceOf[AnyRef])
}
rowsBuffer.add(line)
if (rowsBuffer.size > maxRowCount) {
if (!partitionTaskAtomicity && rowsBuffer.size > maxRowCount) {
flush
}
})
flush
Copy link
Member

@JNSimba JNSimba May 26, 2023

Choose a reason for hiding this comment

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

What does this flush deal with？

Copy link
Author

@baishaoisde baishaoisde May 29, 2023

Choose a reason for hiding this comment

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

This flush is to allow each partition to commit only one StreamLoad task when the partitionTaskAtomicity parameter is true, but seems redundant because the code already has the "if (! rowArray.isEmpty) {flush} "operation.

// flush buffer
if (!rowsBuffer.isEmpty) {
flush
Expand Down
5 changes: 4 additions & 1 deletion spark-doris-connector/src/main/scala/org/apache/doris/spark/sql/DorisStreamLoadSink.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ private[sql] class DorisStreamLoadSink(sqlContext: SQLContext, settings: SparkSe
val maxRetryTimes: Int = settings.getIntegerProperty(ConfigurationOptions.DORIS_SINK_MAX_RETRIES, ConfigurationOptions.SINK_MAX_RETRIES_DEFAULT)
val sinkTaskPartitionSize = settings.getIntegerProperty(ConfigurationOptions.DORIS_SINK_TASK_PARTITION_SIZE)
val sinkTaskUseRepartition = settings.getProperty(ConfigurationOptions.DORIS_SINK_TASK_USE_REPARTITION, ConfigurationOptions.DORIS_SINK_TASK_USE_REPARTITION_DEFAULT.toString).toBoolean
val partitionTaskAtomicity = settings.getProperty(ConfigurationOptions.DORIS_SINK_PER_PARTITION_TASK_ATOMICITY, ConfigurationOptions.DORIS_SINK_PER_PARTITION_TASK_ATOMICITY_DEFAULT.toString).toBoolean
val dorisStreamLoader: DorisStreamLoad = CachedDorisStreamLoadClient.getOrCreate(settings)

override def addBatch(batchId: Long, data: DataFrame): Unit = {
Expand Down Expand Up @@ -70,10 +71,12 @@ private[sql] class DorisStreamLoadSink(sqlContext: SQLContext, settings: SparkSe
}
}
rowArray.add(rowNode)
if (rowArray.size > maxRowCount - 1) {
if (!partitionTaskAtomicity && rowArray.size > maxRowCount - 1) {
flush
}
})
flush

// flush buffer
if (!rowArray.isEmpty) {
flush
Expand Down