You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ge...@apache.org on 2022/12/22 05:39:40 UTC
[spark] branch master updated: [SPARK-41428][UI] Protobuf serializer for SpeculationStageSummaryWrapper
This is an automated email from the ASF dual-hosted git repository.
gengliang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 782a9dfe7b0 [SPARK-41428][UI] Protobuf serializer for SpeculationStageSummaryWrapper
782a9dfe7b0 is described below
commit 782a9dfe7b0e284d4161e3b2cd3c496ea3d16765
Author: Sandeep Singh <sa...@techaddict.me>
AuthorDate: Wed Dec 21 21:39:27 2022 -0800
[SPARK-41428][UI] Protobuf serializer for SpeculationStageSummaryWrapper
### What changes were proposed in this pull request?
Add Protobuf serializer for SpeculationStageSummaryWrapper
### Why are the changes needed?
Support fast and compact serialization/deserialization for SpeculationStageSummaryWrapper over RocksDB.
### Does this PR introduce any user-facing change?
No
### How was this patch tested?
New UT
Closes #39108 from techaddict/SPARK-41428-SpeculationStageSummaryWrapper.
Authored-by: Sandeep Singh <sa...@techaddict.me>
Signed-off-by: Gengliang Wang <ge...@apache.org>
---
.../apache/spark/status/protobuf/store_types.proto | 14 +++++
.../org.apache.spark.status.protobuf.ProtobufSerDe | 1 +
.../SpeculationStageSummaryWrapperSerializer.scala | 68 ++++++++++++++++++++++
.../protobuf/KVStoreProtobufSerializerSuite.scala | 23 ++++++++
4 files changed, 106 insertions(+)
diff --git a/core/src/main/protobuf/org/apache/spark/status/protobuf/store_types.proto b/core/src/main/protobuf/org/apache/spark/status/protobuf/store_types.proto
index d406add82e4..9be0f40fa02 100644
--- a/core/src/main/protobuf/org/apache/spark/status/protobuf/store_types.proto
+++ b/core/src/main/protobuf/org/apache/spark/status/protobuf/store_types.proto
@@ -244,3 +244,17 @@ message RDDStorageInfoWrapper {
message ResourceProfileWrapper {
ResourceProfileInfo rpInfo = 1;
}
+
+message SpeculationStageSummary {
+ int32 num_tasks = 1;
+ int32 num_active_tasks = 2;
+ int32 num_completed_tasks = 3;
+ int32 num_failed_tasks = 4;
+ int32 num_killed_tasks = 5;
+}
+
+message SpeculationStageSummaryWrapper {
+ int64 stage_id = 1;
+ int32 stage_attempt_id = 2;
+ SpeculationStageSummary info = 3;
+}
diff --git a/core/src/main/resources/META-INF/services/org.apache.spark.status.protobuf.ProtobufSerDe b/core/src/main/resources/META-INF/services/org.apache.spark.status.protobuf.ProtobufSerDe
index 4e09700981d..c7aedebfe3e 100644
--- a/core/src/main/resources/META-INF/services/org.apache.spark.status.protobuf.ProtobufSerDe
+++ b/core/src/main/resources/META-INF/services/org.apache.spark.status.protobuf.ProtobufSerDe
@@ -22,3 +22,4 @@ org.apache.spark.status.protobuf.ExecutorStageSummaryWrapperSerializer
org.apache.spark.status.protobuf.TaskDataWrapperSerializer
org.apache.spark.status.protobuf.JobDataWrapperSerializer
org.apache.spark.status.protobuf.ResourceProfileWrapperSerializer
+org.apache.spark.status.protobuf.SpeculationStageSummaryWrapperSerializer
diff --git a/core/src/main/scala/org/apache/spark/status/protobuf/SpeculationStageSummaryWrapperSerializer.scala b/core/src/main/scala/org/apache/spark/status/protobuf/SpeculationStageSummaryWrapperSerializer.scala
new file mode 100644
index 00000000000..936e2dd4504
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/status/protobuf/SpeculationStageSummaryWrapperSerializer.scala
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.status.protobuf
+
+import org.apache.spark.status.SpeculationStageSummaryWrapper
+import org.apache.spark.status.api.v1.SpeculationStageSummary
+
+class SpeculationStageSummaryWrapperSerializer extends ProtobufSerDe {
+
+ override val supportClass: Class[_] = classOf[SpeculationStageSummaryWrapper]
+
+ override def serialize(input: Any): Array[Byte] =
+ serialize(input.asInstanceOf[SpeculationStageSummaryWrapper])
+
+ private def serialize(s: SpeculationStageSummaryWrapper): Array[Byte] = {
+ val summary = serializeSpeculationStageSummary(s.info)
+ val builder = StoreTypes.SpeculationStageSummaryWrapper.newBuilder()
+ builder.setStageId(s.stageId.toLong)
+ builder.setStageAttemptId(s.stageAttemptId)
+ builder.setInfo(summary)
+ builder.build().toByteArray
+ }
+
+ def deserialize(bytes: Array[Byte]): SpeculationStageSummaryWrapper = {
+ val wrapper = StoreTypes.SpeculationStageSummaryWrapper.parseFrom(bytes)
+ new SpeculationStageSummaryWrapper(
+ stageId = wrapper.getStageId.toInt,
+ stageAttemptId = wrapper.getStageAttemptId,
+ info = deserializeSpeculationStageSummary(wrapper.getInfo)
+ )
+ }
+
+ private def serializeSpeculationStageSummary(summary: SpeculationStageSummary):
+ StoreTypes.SpeculationStageSummary = {
+ val summaryBuilder = StoreTypes.SpeculationStageSummary.newBuilder()
+ summaryBuilder.setNumTasks(summary.numTasks)
+ summaryBuilder.setNumActiveTasks(summary.numActiveTasks)
+ summaryBuilder.setNumCompletedTasks(summary.numCompletedTasks)
+ summaryBuilder.setNumFailedTasks(summary.numFailedTasks)
+ summaryBuilder.setNumKilledTasks(summary.numKilledTasks)
+ summaryBuilder.build()
+ }
+
+ private def deserializeSpeculationStageSummary(info: StoreTypes.SpeculationStageSummary):
+ SpeculationStageSummary = {
+ new SpeculationStageSummary(
+ numTasks = info.getNumTasks,
+ numActiveTasks = info.getNumActiveTasks,
+ numCompletedTasks = info.getNumCompletedTasks,
+ numFailedTasks = info.getNumFailedTasks,
+ numKilledTasks = info.getNumKilledTasks)
+ }
+}
diff --git a/core/src/test/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializerSuite.scala b/core/src/test/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializerSuite.scala
index 996a4f0d700..cc6e03ba87b 100644
--- a/core/src/test/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializerSuite.scala
@@ -500,4 +500,27 @@ class KVStoreProtobufSerializerSuite extends SparkFunSuite {
assert(result.rpInfo.taskResources(k) == input.rpInfo.taskResources(k))
}
}
+
+ test("Speculation Stage Summary") {
+ val input = new SpeculationStageSummaryWrapper(
+ stageId = 1,
+ stageAttemptId = 2,
+ info = new SpeculationStageSummary(
+ numTasks = 3,
+ numActiveTasks = 4,
+ numCompletedTasks = 5,
+ numFailedTasks = 6,
+ numKilledTasks = 7
+ )
+ )
+ val bytes = serializer.serialize(input)
+ val result = serializer.deserialize(bytes, classOf[SpeculationStageSummaryWrapper])
+ assert(result.stageId == input.stageId)
+ assert(result.stageAttemptId == input.stageAttemptId)
+ assert(result.info.numTasks == input.info.numTasks)
+ assert(result.info.numActiveTasks == input.info.numActiveTasks)
+ assert(result.info.numCompletedTasks == input.info.numCompletedTasks)
+ assert(result.info.numFailedTasks == input.info.numFailedTasks)
+ assert(result.info.numKilledTasks == input.info.numKilledTasks)
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org