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/17 01:58:30 UTC
[spark] branch master updated: [SPARK-41421][UI] Protobuf serializer for ApplicationEnvironmentInfoWrapper
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 45d9daa2ecf [SPARK-41421][UI] Protobuf serializer for ApplicationEnvironmentInfoWrapper
45d9daa2ecf is described below
commit 45d9daa2ecf6081ef1d031065a9c0e9a3a7f7a58
Author: Sandeep Singh <sa...@techaddict.me>
AuthorDate: Fri Dec 16 17:58:10 2022 -0800
[SPARK-41421][UI] Protobuf serializer for ApplicationEnvironmentInfoWrapper
### What changes were proposed in this pull request?
Add Protobuf serializer for ApplicationEnvironmentInfoWrapper
### Why are the changes needed?
Support fast and compact serialization/deserialization for ApplicationEnvironmentInfoWrapper over RocksDB.
### Does this PR introduce any user-facing change?
No
### How was this patch tested?
New UT
Closes #39096 from techaddict/SPARK-41421-ApplicationEnvironmentInfoWrapper.
Authored-by: Sandeep Singh <sa...@techaddict.me>
Signed-off-by: Gengliang Wang <ge...@apache.org>
---
.../apache/spark/status/protobuf/store_types.proto | 43 ++++++
...plicationEnvironmentInfoWrapperSerializer.scala | 146 +++++++++++++++++++++
.../protobuf/KVStoreProtobufSerializer.scala | 6 +-
.../protobuf/KVStoreProtobufSerializerSuite.scala | 84 +++++++++++-
4 files changed, 276 insertions(+), 3 deletions(-)
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 23cb908a691..1ee22028c04 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
@@ -106,3 +106,46 @@ message TaskDataWrapper {
int64 stage_id = 40;
int32 stage_attempt_id = 41;
}
+
+message ExecutorResourceRequest {
+ string resource_name = 1;
+ int64 amount = 2;
+ string discoveryScript = 3;
+ string vendor = 4;
+}
+
+message TaskResourceRequest {
+ string resource_name = 1;
+ double amount = 2;
+}
+
+message ResourceProfileInfo {
+ int32 id = 1;
+ map<string, ExecutorResourceRequest> executor_resources = 2;
+ map<string, TaskResourceRequest> task_resources = 3;
+}
+
+message RuntimeInfo {
+ string java_version = 1;
+ string java_home = 2;
+ string scala_version = 3;
+}
+
+message PairStrings {
+ string value1 = 1;
+ string value2 = 2;
+}
+
+message ApplicationEnvironmentInfo {
+ RuntimeInfo runtime = 1;
+ repeated PairStrings spark_properties = 2;
+ repeated PairStrings hadoop_properties = 3;
+ repeated PairStrings system_properties = 4;
+ repeated PairStrings metrics_properties = 5;
+ repeated PairStrings classpath_entries = 6;
+ repeated ResourceProfileInfo resource_profiles = 7;
+}
+
+message ApplicationEnvironmentInfoWrapper {
+ ApplicationEnvironmentInfo info = 1;
+}
diff --git a/core/src/main/scala/org/apache/spark/status/protobuf/ApplicationEnvironmentInfoWrapperSerializer.scala b/core/src/main/scala/org/apache/spark/status/protobuf/ApplicationEnvironmentInfoWrapperSerializer.scala
new file mode 100644
index 00000000000..1649200b94a
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/status/protobuf/ApplicationEnvironmentInfoWrapperSerializer.scala
@@ -0,0 +1,146 @@
+/*
+ * 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 collection.JavaConverters._
+
+import org.apache.spark.resource.{ExecutorResourceRequest, TaskResourceRequest}
+import org.apache.spark.status.ApplicationEnvironmentInfoWrapper
+import org.apache.spark.status.api.v1.{ApplicationEnvironmentInfo, ResourceProfileInfo, RuntimeInfo}
+
+object ApplicationEnvironmentInfoWrapperSerializer {
+ def serialize(input: ApplicationEnvironmentInfoWrapper): Array[Byte] = {
+ val builder = StoreTypes.ApplicationEnvironmentInfoWrapper.newBuilder()
+ builder.setInfo(serializeApplicationEnvironmentInfo(input.info))
+ builder.build().toByteArray
+ }
+
+ def deserialize(bytes: Array[Byte]): ApplicationEnvironmentInfoWrapper = {
+ val wrapper = StoreTypes.ApplicationEnvironmentInfoWrapper.parseFrom(bytes)
+ new ApplicationEnvironmentInfoWrapper(
+ info = deserializeApplicationEnvironmentInfo(wrapper.getInfo)
+ )
+ }
+
+ private def serializeApplicationEnvironmentInfo(info: ApplicationEnvironmentInfo):
+ StoreTypes.ApplicationEnvironmentInfo = {
+
+ val runtimeBuilder = StoreTypes.RuntimeInfo.newBuilder()
+ runtimeBuilder.setJavaVersion(info.runtime.javaVersion)
+ runtimeBuilder.setJavaHome(info.runtime.javaHome)
+ runtimeBuilder.setScalaVersion(info.runtime.scalaVersion)
+
+ val builder = StoreTypes.ApplicationEnvironmentInfo.newBuilder()
+ builder.setRuntime(runtimeBuilder.build())
+ info.sparkProperties.foreach { pair =>
+ builder.addSparkProperties(serializePairStrings(pair))
+ }
+ info.hadoopProperties.foreach { pair =>
+ builder.addHadoopProperties(serializePairStrings(pair))
+ }
+ info.systemProperties.foreach { pair =>
+ builder.addSystemProperties(serializePairStrings(pair))
+ }
+ info.metricsProperties.foreach { pair =>
+ builder.addMetricsProperties(serializePairStrings(pair))
+ }
+ info.classpathEntries.foreach { pair =>
+ builder.addClasspathEntries(serializePairStrings(pair))
+ }
+ info.resourceProfiles.foreach { profile =>
+ builder.addResourceProfiles(serializeResourceProfileInfo(profile))
+ }
+ builder.build()
+ }
+
+ private def deserializeApplicationEnvironmentInfo(info: StoreTypes.ApplicationEnvironmentInfo):
+ ApplicationEnvironmentInfo = {
+ val runtime = new RuntimeInfo (
+ javaVersion = info.getRuntime.getJavaVersion,
+ javaHome = info.getRuntime.getJavaHome,
+ scalaVersion = info.getRuntime.getScalaVersion
+ )
+ val pairSSToTuple = (pair: StoreTypes.PairStrings) => {
+ (pair.getValue1, pair.getValue2)
+ }
+ new ApplicationEnvironmentInfo(
+ runtime = runtime,
+ sparkProperties = info.getSparkPropertiesList.asScala.map(pairSSToTuple).toSeq,
+ hadoopProperties = info.getHadoopPropertiesList.asScala.map(pairSSToTuple).toSeq,
+ systemProperties = info.getSystemPropertiesList.asScala.map(pairSSToTuple).toSeq,
+ metricsProperties = info.getMetricsPropertiesList.asScala.map(pairSSToTuple).toSeq,
+ classpathEntries = info.getClasspathEntriesList.asScala.map(pairSSToTuple).toSeq,
+ resourceProfiles =
+ info.getResourceProfilesList.asScala.map(deserializeResourceProfileInfo).toSeq
+ )
+ }
+
+ private def serializePairStrings(pair: (String, String)): StoreTypes.PairStrings = {
+ val builder = StoreTypes.PairStrings.newBuilder()
+ builder.setValue1(pair._1)
+ builder.setValue2(pair._2)
+ builder.build()
+ }
+
+ private def serializeResourceProfileInfo(info: ResourceProfileInfo):
+ StoreTypes.ResourceProfileInfo = {
+ val builder = StoreTypes.ResourceProfileInfo.newBuilder()
+ builder.setId(info.id)
+ info.executorResources.foreach{case (k, resource) =>
+ val requestBuilder = StoreTypes.ExecutorResourceRequest.newBuilder()
+ requestBuilder.setResourceName(resource.resourceName)
+ requestBuilder.setAmount(resource.amount)
+ requestBuilder.setDiscoveryScript(resource.discoveryScript)
+ requestBuilder.setVendor(resource.vendor)
+ builder.putExecutorResources(k, requestBuilder.build())
+ }
+ info.taskResources.foreach { case (k, resource) =>
+ val requestBuilder = StoreTypes.TaskResourceRequest.newBuilder()
+ requestBuilder.setResourceName(resource.resourceName)
+ requestBuilder.setAmount(resource.amount)
+ builder.putTaskResources(k, requestBuilder.build())
+ }
+ builder.build()
+ }
+
+ private def deserializeResourceProfileInfo(info: StoreTypes.ResourceProfileInfo):
+ ResourceProfileInfo = {
+
+ new ResourceProfileInfo(
+ id = info.getId,
+ executorResources =
+ info.getExecutorResourcesMap.asScala.mapValues(deserializeExecutorResourceRequest).toMap,
+ taskResources =
+ info.getTaskResourcesMap.asScala.mapValues(deserializeTaskResourceRequest).toMap)
+ }
+
+ private def deserializeExecutorResourceRequest(info: StoreTypes.ExecutorResourceRequest):
+ ExecutorResourceRequest = {
+ new ExecutorResourceRequest(
+ resourceName = info.getResourceName,
+ amount = info.getAmount,
+ discoveryScript = info.getDiscoveryScript,
+ vendor = info.getVendor
+ )
+ }
+
+ private def deserializeTaskResourceRequest(info: StoreTypes.TaskResourceRequest):
+ TaskResourceRequest = {
+ new TaskResourceRequest(resourceName = info.getResourceName, amount = info.getAmount)
+ }
+}
diff --git a/core/src/main/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializer.scala b/core/src/main/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializer.scala
index b3b896f58f1..fc5eb203938 100644
--- a/core/src/main/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializer.scala
+++ b/core/src/main/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializer.scala
@@ -17,13 +17,15 @@
package org.apache.spark.status.protobuf
-import org.apache.spark.status.{JobDataWrapper, TaskDataWrapper}
+import org.apache.spark.status.{ApplicationEnvironmentInfoWrapper, JobDataWrapper, TaskDataWrapper}
import org.apache.spark.status.KVUtils.KVStoreScalaSerializer
private[spark] class KVStoreProtobufSerializer extends KVStoreScalaSerializer {
override def serialize(o: Object): Array[Byte] = o match {
case j: JobDataWrapper => JobDataWrapperSerializer.serialize(j)
case t: TaskDataWrapper => TaskDataWrapperSerializer.serialize(t)
+ case a: ApplicationEnvironmentInfoWrapper =>
+ ApplicationEnvironmentInfoWrapperSerializer.serialize(a)
case other => super.serialize(other)
}
@@ -32,6 +34,8 @@ private[spark] class KVStoreProtobufSerializer extends KVStoreScalaSerializer {
JobDataWrapperSerializer.deserialize(data).asInstanceOf[T]
case _ if classOf[TaskDataWrapper].isAssignableFrom(klass) =>
TaskDataWrapperSerializer.deserialize(data).asInstanceOf[T]
+ case _ if classOf[ApplicationEnvironmentInfoWrapper].isAssignableFrom(klass) =>
+ ApplicationEnvironmentInfoWrapperSerializer.deserialize(data).asInstanceOf[T]
case other => super.deserialize(data, klass)
}
}
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 f35a28d050c..8ee1126bd68 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
@@ -20,8 +20,9 @@ package org.apache.spark.status.protobuf
import java.util.Date
import org.apache.spark.{JobExecutionStatus, SparkFunSuite}
-import org.apache.spark.status.{JobDataWrapper, TaskDataWrapper}
-import org.apache.spark.status.api.v1.{AccumulableInfo, JobData}
+import org.apache.spark.resource.{ExecutorResourceRequest, TaskResourceRequest}
+import org.apache.spark.status.{ApplicationEnvironmentInfoWrapper, JobDataWrapper, TaskDataWrapper}
+import org.apache.spark.status.api.v1.{AccumulableInfo, ApplicationEnvironmentInfo, JobData, ResourceProfileInfo, RuntimeInfo}
class KVStoreProtobufSerializerSuite extends SparkFunSuite {
private val serializer = new KVStoreProtobufSerializer()
@@ -176,4 +177,83 @@ class KVStoreProtobufSerializerSuite extends SparkFunSuite {
assert(result.stageId == input.stageId)
assert(result.stageAttemptId == input.stageAttemptId)
}
+
+ test("Application Environment Info") {
+ val input = new ApplicationEnvironmentInfoWrapper(
+ new ApplicationEnvironmentInfo(
+ runtime = new RuntimeInfo(
+ javaVersion = "1.8",
+ javaHome = "/tmp/java",
+ scalaVersion = "2.13"),
+ sparkProperties = Seq(("spark.conf.1", "1"), ("spark.conf.2", "2")),
+ hadoopProperties = Seq(("hadoop.conf.conf1", "1"), ("hadoop.conf2", "val2")),
+ systemProperties = Seq(("sys.prop.1", "value1"), ("sys.prop.2", "value2")),
+ metricsProperties = Seq(("metric.1", "klass1"), ("metric2", "klass2")),
+ classpathEntries = Seq(("/jar1", "System"), ("/jar2", "User")),
+ resourceProfiles = Seq(new ResourceProfileInfo(
+ id = 0,
+ executorResources = Map(
+ "0" -> new ExecutorResourceRequest(
+ resourceName = "exec1",
+ amount = 1,
+ discoveryScript = "script0",
+ vendor = "apache"),
+ "1" -> new ExecutorResourceRequest(
+ resourceName = "exec2",
+ amount = 1,
+ discoveryScript = "script1",
+ vendor = "apache")
+ ),
+ taskResources = Map(
+ "0" -> new TaskResourceRequest(resourceName = "exec1", amount = 1),
+ "1" -> new TaskResourceRequest(resourceName = "exec2", amount = 1)
+ )
+ ))
+ )
+ )
+
+ val bytes = serializer.serialize(input)
+ val result = serializer.deserialize(bytes, classOf[ApplicationEnvironmentInfoWrapper])
+ assert(result.info.runtime.javaVersion == input.info.runtime.javaVersion)
+ assert(result.info.runtime.javaHome == input.info.runtime.javaHome)
+ assert(result.info.runtime.scalaVersion == input.info.runtime.scalaVersion)
+ assert(result.info.sparkProperties.length == input.info.sparkProperties.length)
+ result.info.sparkProperties.zip(input.info.sparkProperties).foreach { case (p1, p2) =>
+ assert(p1 == p2)
+ }
+ assert(result.info.hadoopProperties.length == input.info.hadoopProperties.length)
+ result.info.hadoopProperties.zip(input.info.hadoopProperties).foreach { case (p1, p2) =>
+ assert(p1 == p2)
+ }
+ assert(result.info.systemProperties.length == input.info.systemProperties.length)
+ result.info.systemProperties.zip(input.info.systemProperties).foreach { case (p1, p2) =>
+ assert(p1 == p2)
+ }
+ assert(result.info.metricsProperties.length == input.info.metricsProperties.length)
+ result.info.metricsProperties.zip(input.info.metricsProperties).foreach { case (p1, p2) =>
+ assert(p1 == p2)
+ }
+ assert(result.info.classpathEntries.length == input.info.classpathEntries.length)
+ result.info.classpathEntries.zip(input.info.classpathEntries).foreach { case (p1, p2) =>
+ assert(p1 == p2)
+ }
+ assert(result.info.resourceProfiles.length == input.info.resourceProfiles.length)
+ result.info.resourceProfiles.zip(input.info.resourceProfiles).foreach { case (p1, p2) =>
+ assert(p1.id == p2.id)
+ assert(p1.executorResources.size == p2.executorResources.size)
+ assert(p1.executorResources.keys.size == p2.executorResources.keys.size)
+ p1.executorResources.keysIterator.foreach { k =>
+ assert(p1.executorResources.contains(k))
+ assert(p2.executorResources.contains(k))
+ assert(p1.executorResources(k) == p2.executorResources(k))
+ }
+ assert(p1.taskResources.size == p2.taskResources.size)
+ assert(p1.taskResources.keys.size == p2.taskResources.keys.size)
+ p1.taskResources.keysIterator.foreach { k =>
+ assert(p1.taskResources.contains(k))
+ assert(p2.taskResources.contains(k))
+ assert(p1.taskResources(k) == p2.taskResources(k))
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org