You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2024/06/30 05:04:33 UTC
(paimon) branch master updated: [spark] Optimize sort compact procedure by submitting job by partition (#3613)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new e2d89c443 [spark] Optimize sort compact procedure by submitting job by partition (#3613)
e2d89c443 is described below
commit e2d89c4436025f4ff94222a41e295995dd9cf410
Author: Zouxxyy <zo...@alibaba-inc.com>
AuthorDate: Sun Jun 30 13:04:28 2024 +0800
[spark] Optimize sort compact procedure by submitting job by partition (#3613)
---
docs/content/spark/procedures.md | 1 +
.../paimon/spark/procedure/BaseProcedure.java | 3 +-
.../paimon/spark/procedure/CompactProcedure.java | 124 ++++++++++++++++-----
.../org/apache/paimon/spark/PaimonSplitScan.scala | 6 +
.../paimon/spark/commands/PaimonSparkWriter.scala | 2 +-
.../spark/procedure/CompactProcedureTestBase.scala | 47 ++++++++
6 files changed, 153 insertions(+), 30 deletions(-)
diff --git a/docs/content/spark/procedures.md b/docs/content/spark/procedures.md
index 2b55bb9f6..da9d7d33d 100644
--- a/docs/content/spark/procedures.md
+++ b/docs/content/spark/procedures.md
@@ -46,6 +46,7 @@ This section introduce all available spark procedures about paimon.
<li>where: partition predicate. Left empty for all partitions. (Can't be used together with "partitions")</li>
<li>order_strategy: 'order' or 'zorder' or 'hilbert' or 'none'. Left empty for 'none'.</li>
<li>order_columns: the columns need to be sort. Left empty if 'order_strategy' is 'none'.</li>
+ <li>max_concurrent_jobs: when sort compact is used, files in one partition are grouped and submitted as a single spark compact job. This parameter controls the maximum number of jobs that can be submitted simultaneously. The default value is 15.</li>
</td>
<td>
SET spark.sql.shuffle.partitions=10; --set the compact parallelism <br/>
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BaseProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BaseProcedure.java
index b506b2fa1..fe9d01971 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BaseProcedure.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/BaseProcedure.java
@@ -26,7 +26,6 @@ import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan;
import org.apache.spark.sql.connector.catalog.CatalogPlugin;
import org.apache.spark.sql.connector.catalog.Identifier;
import org.apache.spark.sql.connector.catalog.Table;
@@ -109,7 +108,7 @@ abstract class BaseProcedure implements Procedure {
}
}
- protected LogicalPlan createRelation(Identifier ident) {
+ protected DataSourceV2Relation createRelation(Identifier ident) {
return DataSourceV2Relation.create(
loadSparkTable(ident), Option.apply(tableCatalog), Option.apply(ident));
}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
index 0f64f95c2..e21702bab 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java
@@ -22,14 +22,15 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.append.AppendOnlyCompactionTask;
import org.apache.paimon.append.AppendOnlyTableCompactionCoordinator;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.operation.AppendOnlyFileStoreWrite;
-import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.spark.DynamicOverWrite$;
+import org.apache.paimon.spark.PaimonSplitScan;
import org.apache.paimon.spark.SparkUtils;
+import org.apache.paimon.spark.catalyst.Compatibility;
import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionUtils;
-import org.apache.paimon.spark.commands.WriteIntoPaimonTable;
+import org.apache.paimon.spark.commands.PaimonSparkWriter;
import org.apache.paimon.spark.sort.TableSorter;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
@@ -42,6 +43,7 @@ import org.apache.paimon.table.sink.CompactionTaskSerializer;
import org.apache.paimon.table.sink.TableCommitImpl;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.utils.ExecutorThreadFactory;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.ParameterUtils;
import org.apache.paimon.utils.SerializationUtils;
@@ -55,14 +57,16 @@ import org.apache.spark.sql.PaimonUtils;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.catalyst.expressions.Expression;
-import org.apache.spark.sql.catalyst.plans.logical.Filter;
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan;
import org.apache.spark.sql.connector.catalog.Identifier;
import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
@@ -71,13 +75,21 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.stream.Collectors;
+import scala.collection.Seq;
+import scala.collection.mutable.ListBuffer;
+
import static org.apache.paimon.CoreOptions.createCommitUser;
import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.spark.sql.types.DataTypes.IntegerType;
import static org.apache.spark.sql.types.DataTypes.StringType;
/**
@@ -89,13 +101,16 @@ import static org.apache.spark.sql.types.DataTypes.StringType;
*/
public class CompactProcedure extends BaseProcedure {
+ private static final Logger LOG = LoggerFactory.getLogger(CompactProcedure.class.getName());
+
private static final ProcedureParameter[] PARAMETERS =
new ProcedureParameter[] {
ProcedureParameter.required("table", StringType),
ProcedureParameter.optional("partitions", StringType),
ProcedureParameter.optional("order_strategy", StringType),
ProcedureParameter.optional("order_by", StringType),
- ProcedureParameter.optional("where", StringType)
+ ProcedureParameter.optional("where", StringType),
+ ProcedureParameter.optional("max_concurrent_jobs", IntegerType)
};
private static final StructType OUTPUT_TYPE =
@@ -128,6 +143,7 @@ public class CompactProcedure extends BaseProcedure {
? Collections.emptyList()
: Arrays.asList(args.getString(3).split(","));
String where = blank(args, 4) ? null : args.getString(4);
+ int maxConcurrentJobs = args.isNullAt(5) ? 15 : args.getInt(5);
if (TableSorter.OrderType.NONE.name().equals(sortType) && !sortColumns.isEmpty()) {
throw new IllegalArgumentException(
"order_strategy \"none\" cannot work with order_by columns.");
@@ -140,7 +156,12 @@ public class CompactProcedure extends BaseProcedure {
tableIdent,
table -> {
checkArgument(table instanceof FileStoreTable);
- LogicalPlan relation = createRelation(tableIdent);
+ checkArgument(
+ sortColumns.stream().noneMatch(table.partitionKeys()::contains),
+ "order_by should not contain partition cols, because it is meaningless, your order_by cols are %s, and partition cols are %s",
+ sortColumns,
+ table.partitionKeys());
+ DataSourceV2Relation relation = createRelation(tableIdent);
Expression condition = null;
if (!StringUtils.isBlank(finalWhere)) {
condition = ExpressionUtils.resolveFilter(spark(), relation, finalWhere);
@@ -160,7 +181,8 @@ public class CompactProcedure extends BaseProcedure {
sortType,
sortColumns,
relation,
- condition));
+ condition,
+ maxConcurrentJobs));
return new InternalRow[] {internalRow};
});
}
@@ -178,19 +200,23 @@ public class CompactProcedure extends BaseProcedure {
FileStoreTable table,
String sortType,
List<String> sortColumns,
- LogicalPlan relation,
- @Nullable Expression condition) {
+ DataSourceV2Relation relation,
+ @Nullable Expression condition,
+ int maxConcurrentJobs) {
table = table.copy(Collections.singletonMap(CoreOptions.WRITE_ONLY.key(), "false"));
BucketMode bucketMode = table.bucketMode();
TableSorter.OrderType orderType = TableSorter.OrderType.of(sortType);
+ Predicate filter =
+ condition == null
+ ? null
+ : ExpressionUtils.convertConditionToPaimonPredicate(
+ condition,
+ ((LogicalPlan) relation).output(),
+ table.rowType(),
+ false)
+ .getOrElse(null);
if (orderType.equals(TableSorter.OrderType.NONE)) {
JavaSparkContext javaSparkContext = new JavaSparkContext(spark().sparkContext());
- Predicate filter =
- condition == null
- ? null
- : ExpressionUtils.convertConditionToPaimonPredicate(
- condition, relation.output(), table.rowType(), false)
- .getOrElse(null);
switch (bucketMode) {
case HASH_FIXED:
case HASH_DYNAMIC:
@@ -207,7 +233,7 @@ public class CompactProcedure extends BaseProcedure {
switch (bucketMode) {
case BUCKET_UNAWARE:
sortCompactUnAwareBucketTable(
- table, orderType, sortColumns, relation, condition);
+ table, orderType, sortColumns, relation, filter, maxConcurrentJobs);
break;
default:
throw new UnsupportedOperationException(
@@ -360,17 +386,61 @@ public class CompactProcedure extends BaseProcedure {
FileStoreTable table,
TableSorter.OrderType orderType,
List<String> sortColumns,
- LogicalPlan relation,
- @Nullable Expression condition) {
- Dataset<Row> row =
- PaimonUtils.createDataset(
- spark(), condition == null ? relation : new Filter(condition, relation));
- new WriteIntoPaimonTable(
- table,
- DynamicOverWrite$.MODULE$,
- TableSorter.getSorter(table, orderType, sortColumns).sort(row),
- new Options())
- .run(spark());
+ DataSourceV2Relation relation,
+ @Nullable Predicate filter,
+ int maxConcurrentJobs) {
+ SnapshotReader snapshotReader = table.newSnapshotReader();
+ if (filter != null) {
+ snapshotReader.withFilter(filter);
+ }
+ Map<BinaryRow, DataSplit[]> packedSplits = packForSort(snapshotReader.read().dataSplits());
+
+ PaimonSparkWriter writer = new PaimonSparkWriter(table);
+ // Use dynamic partition overwrite
+ writer.writeBuilder().withOverwrite();
+
+ ExecutorService executorService =
+ Executors.newFixedThreadPool(
+ Math.min(maxConcurrentJobs, packedSplits.size()),
+ new ExecutorThreadFactory(Thread.currentThread().getName() + "-compact"));
+ LinkedList<Future<Seq<CommitMessage>>> futures = new LinkedList<>();
+ TableSorter sorter = TableSorter.getSorter(table, orderType, sortColumns);
+
+ LOG.info("Start submit sort compact jobs, count: {}", packedSplits.size());
+ for (Map.Entry<BinaryRow, DataSplit[]> entry : packedSplits.entrySet()) {
+ Dataset<Row> dataset =
+ PaimonUtils.createDataset(
+ spark(),
+ Compatibility.createDataSourceV2ScanRelation(
+ relation,
+ PaimonSplitScan.apply(table, entry.getValue()),
+ relation.output()));
+ futures.add(executorService.submit(() -> writer.write(sorter.sort(dataset))));
+ }
+
+ ListBuffer<CommitMessage> messages = new ListBuffer<>();
+ try {
+ for (Future<Seq<CommitMessage>> future : futures) {
+ messages.append(future.get());
+ }
+ } catch (Throwable e) {
+ throw new RuntimeException("Compact failed", e);
+ } finally {
+ executorService.shutdownNow();
+ }
+
+ writer.commit(messages.toSeq());
+ }
+
+ private Map<BinaryRow, DataSplit[]> packForSort(List<DataSplit> dataSplits) {
+ // Make a single partition as a compact group
+ return dataSplits.stream()
+ .collect(
+ Collectors.groupingBy(
+ DataSplit::partition,
+ Collectors.collectingAndThen(
+ Collectors.toList(),
+ list -> list.toArray(new DataSplit[0]))));
}
@VisibleForTesting
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSplitScan.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSplitScan.scala
index 58c2fd693..9f792de48 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSplitScan.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSplitScan.scala
@@ -45,3 +45,9 @@ case class PaimonSplitScan(
metadataColumns)
}
}
+
+object PaimonSplitScan {
+ def apply(table: Table, dataSplits: Array[DataSplit]): PaimonSplitScan = {
+ new PaimonSplitScan(table, dataSplits)
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
index 7be4dd80a..41f2db1a3 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
@@ -50,7 +50,7 @@ case class PaimonSparkWriter(table: FileStoreTable) {
private lazy val primaryKeyCols = tableSchema.trimmedPrimaryKeys().asScala
- private lazy val serializer = new CommitMessageSerializer
+ @transient private lazy val serializer = new CommitMessageSerializer
val writeBuilder: BatchWriteBuilder = table.newBatchWriteBuilder()
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
index 3c5b02605..245076f7b 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala
@@ -30,6 +30,8 @@ import org.assertj.core.api.Assertions
import java.util
+import scala.collection.JavaConverters._
+
/** Test compact procedure. See [[CompactProcedure]]. */
abstract class CompactProcedureTestBase extends PaimonSparkTestBase with StreamTest {
@@ -247,6 +249,47 @@ abstract class CompactProcedureTestBase extends PaimonSparkTestBase with StreamT
}
}
+ test("Paimon Procedure: sort compact with max_concurrent_jobs") {
+ Seq("order", "zorder").foreach {
+ orderStrategy =>
+ {
+ withTable("T") {
+ spark.sql(s"""
+ |CREATE TABLE T (id INT, pt STRING)
+ |PARTITIONED BY (pt)
+ |""".stripMargin)
+
+ spark.sql(s"""INSERT INTO T VALUES
+ |(1, 'p1'), (3, 'p1'),
+ |(1, 'p2'), (4, 'p2'),
+ |(3, 'p3'), (2, 'p3'),
+ |(1, 'p4'), (2, 'p4')
+ |""".stripMargin)
+
+ spark.sql(s"""INSERT INTO T VALUES
+ |(4, 'p1'), (2, 'p1'),
+ |(2, 'p2'), (3, 'p2'),
+ |(1, 'p3'), (4, 'p3'),
+ |(3, 'p4'), (4, 'p4')
+ |""".stripMargin)
+
+ checkAnswer(
+ spark.sql(
+ s"CALL sys.compact(table => 'T', order_strategy => '$orderStrategy', order_by => 'id', max_concurrent_jobs => 2)"),
+ Seq(true).toDF())
+
+ val result = List(Row(1), Row(2), Row(3), Row(4)).asJava
+ Seq("p1", "p2", "p3", "p4").foreach {
+ pt =>
+ Assertions
+ .assertThat(spark.sql(s"SELECT id FROM T WHERE pt='$pt'").collect())
+ .containsExactlyElementsOf(result)
+ }
+ }
+ }
+ }
+ }
+
test("Paimon Procedure: compact for pk") {
failAfter(streamingTimeout) {
withTempDir {
@@ -433,6 +476,10 @@ abstract class CompactProcedureTestBase extends PaimonSparkTestBase with StreamT
assert(intercept[IllegalArgumentException] {
spark.sql("CALL sys.compact(table => 'T', where => 'id > 1 AND pt = \"p1\"')")
}.getMessage.contains("Only partition predicate is supported"))
+
+ assert(intercept[IllegalArgumentException] {
+ spark.sql("CALL sys.compact(table => 'T', order_strategy => 'sort', order_by => 'pt')")
+ }.getMessage.contains("order_by should not contain partition cols"))
}
test("Paimon Procedure: compact with where") {