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 55f64cfa839 [SPARK-41968][CORE][SQL] Refactor `ProtobufSerDe` to `ProtobufSerDe[T]` 55f64cfa839 is described below commit 55f64cfa839a77b8fff7c1625281b84cce4c6807 Author: yangjie01 <yangji...@baidu.com> AuthorDate: Wed Jan 11 22:21:21 2023 -0800 [SPARK-41968][CORE][SQL] Refactor `ProtobufSerDe` to `ProtobufSerDe[T]` ### What changes were proposed in this pull request? This pr aims refator `ProtobufSerDe` to `ProtobufSerDe[T]`, the main change of `ProtobufSerDe` as follows: - Change the definition of `ProtobufSerDe` to `ProtobufSerDe[T]` - Remove `supportClass` method from `ProtobufSerDe[T]` and use reflection in `KVStoreProtobufSerializer` to obtain the actual type of `T` as `serializerMap` key - Change the input parameter type of `serialize` function from `Any` to `T` - Change the return value type of `deserialize` function method from `Any` to `T` Then, all the subclasses of `ProtobufSerDe[T]` are refactored and code cleaned in this pr. ### Why are the changes needed? Refactor `ProtobufSerDe` and code cleanup. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Pass Github Actions Closes #39487 from LuciferYang/refactor-ProtobufSerDe. Authored-by: yangjie01 <yangji...@baidu.com> Signed-off-by: Gengliang Wang <gengli...@apache.org> --- .../ApplicationEnvironmentInfoWrapperSerializer.scala | 10 +++------- .../protobuf/ApplicationInfoWrapperSerializer.scala | 9 ++------- .../status/protobuf/CachedQuantileSerializer.scala | 6 ++---- .../ExecutorStageSummaryWrapperSerializer.scala | 10 +++------- .../protobuf/ExecutorSummaryWrapperSerializer.scala | 10 ++-------- .../status/protobuf/JobDataWrapperSerializer.scala | 9 ++------- .../status/protobuf/KVStoreProtobufSerializer.scala | 15 +++++++++++---- .../protobuf/ProcessSummaryWrapperSerializer.scala | 10 ++-------- .../apache/spark/status/protobuf/ProtobufSerDe.scala | 19 +++++-------------- .../protobuf/RDDOperationGraphWrapperSerializer.scala | 7 ++----- .../protobuf/RDDStorageInfoWrapperSerializer.scala | 9 ++------- .../protobuf/ResourceProfileWrapperSerializer.scala | 9 ++------- .../SpeculationStageSummaryWrapperSerializer.scala | 10 +++------- .../status/protobuf/StageDataWrapperSerializer.scala | 13 +++++-------- .../status/protobuf/StreamBlockDataSerializer.scala | 6 ++---- .../status/protobuf/TaskDataWrapperSerializer.scala | 9 ++------- .../protobuf/sql/SQLExecutionUIDataSerializer.scala | 7 ++----- .../sql/SparkPlanGraphWrapperSerializer.scala | 7 ++----- .../protobuf/sql/StreamingQueryDataSerializer.scala | 9 +++------ 19 files changed, 57 insertions(+), 127 deletions(-) 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 index 33a18daacbc..b7cf01382e2 100644 --- a/core/src/main/scala/org/apache/spark/status/protobuf/ApplicationEnvironmentInfoWrapperSerializer.scala +++ b/core/src/main/scala/org/apache/spark/status/protobuf/ApplicationEnvironmentInfoWrapperSerializer.scala @@ -23,14 +23,10 @@ import org.apache.spark.resource.{ExecutorResourceRequest, TaskResourceRequest} import org.apache.spark.status.ApplicationEnvironmentInfoWrapper import org.apache.spark.status.api.v1.{ApplicationEnvironmentInfo, ResourceProfileInfo, RuntimeInfo} -class ApplicationEnvironmentInfoWrapperSerializer extends ProtobufSerDe { +class ApplicationEnvironmentInfoWrapperSerializer + extends ProtobufSerDe[ApplicationEnvironmentInfoWrapper] { - override val supportClass: Class[_] = classOf[ApplicationEnvironmentInfoWrapper] - - override def serialize(input: Any): Array[Byte] = - serialize(input.asInstanceOf[ApplicationEnvironmentInfoWrapper]) - - private def serialize(input: ApplicationEnvironmentInfoWrapper): Array[Byte] = { + override def serialize(input: ApplicationEnvironmentInfoWrapper): Array[Byte] = { val builder = StoreTypes.ApplicationEnvironmentInfoWrapper.newBuilder() builder.setInfo(serializeApplicationEnvironmentInfo(input.info)) builder.build().toByteArray diff --git a/core/src/main/scala/org/apache/spark/status/protobuf/ApplicationInfoWrapperSerializer.scala b/core/src/main/scala/org/apache/spark/status/protobuf/ApplicationInfoWrapperSerializer.scala index 5a2accb7506..c56b5302cc1 100644 --- a/core/src/main/scala/org/apache/spark/status/protobuf/ApplicationInfoWrapperSerializer.scala +++ b/core/src/main/scala/org/apache/spark/status/protobuf/ApplicationInfoWrapperSerializer.scala @@ -26,14 +26,9 @@ import org.apache.spark.status.api.v1.{ApplicationAttemptInfo, ApplicationInfo} import org.apache.spark.status.protobuf.Utils.getOptional -class ApplicationInfoWrapperSerializer extends ProtobufSerDe { +class ApplicationInfoWrapperSerializer extends ProtobufSerDe[ApplicationInfoWrapper] { - override val supportClass: Class[_] = classOf[ApplicationInfoWrapper] - - override def serialize(input: Any): Array[Byte] = - serialize(input.asInstanceOf[ApplicationInfoWrapper]) - - private def serialize(j: ApplicationInfoWrapper): Array[Byte] = { + override def serialize(j: ApplicationInfoWrapper): Array[Byte] = { val jobData = serializeApplicationInfo(j.info) val builder = StoreTypes.ApplicationInfoWrapper.newBuilder() builder.setInfo(jobData) diff --git a/core/src/main/scala/org/apache/spark/status/protobuf/CachedQuantileSerializer.scala b/core/src/main/scala/org/apache/spark/status/protobuf/CachedQuantileSerializer.scala index 547cbd86b7a..08b8c8b0a98 100644 --- a/core/src/main/scala/org/apache/spark/status/protobuf/CachedQuantileSerializer.scala +++ b/core/src/main/scala/org/apache/spark/status/protobuf/CachedQuantileSerializer.scala @@ -19,11 +19,9 @@ package org.apache.spark.status.protobuf import org.apache.spark.status.CachedQuantile -class CachedQuantileSerializer extends ProtobufSerDe { - override val supportClass: Class[_] = classOf[CachedQuantile] +class CachedQuantileSerializer extends ProtobufSerDe[CachedQuantile] { - override def serialize(input: Any): Array[Byte] = { - val data = input.asInstanceOf[CachedQuantile] + override def serialize(data: CachedQuantile): Array[Byte] = { val builder = StoreTypes.CachedQuantile.newBuilder() .setStageId(data.stageId.toLong) .setStageAttemptId(data.stageAttemptId) diff --git a/core/src/main/scala/org/apache/spark/status/protobuf/ExecutorStageSummaryWrapperSerializer.scala b/core/src/main/scala/org/apache/spark/status/protobuf/ExecutorStageSummaryWrapperSerializer.scala index 71de2fbc81f..4d9d045ed5e 100644 --- a/core/src/main/scala/org/apache/spark/status/protobuf/ExecutorStageSummaryWrapperSerializer.scala +++ b/core/src/main/scala/org/apache/spark/status/protobuf/ExecutorStageSummaryWrapperSerializer.scala @@ -19,14 +19,10 @@ package org.apache.spark.status.protobuf import org.apache.spark.status.ExecutorStageSummaryWrapper -class ExecutorStageSummaryWrapperSerializer extends ProtobufSerDe { +class ExecutorStageSummaryWrapperSerializer + extends ProtobufSerDe[ExecutorStageSummaryWrapper] { - override val supportClass: Class[_] = classOf[ExecutorStageSummaryWrapper] - - override def serialize(input: Any): Array[Byte] = - serialize(input.asInstanceOf[ExecutorStageSummaryWrapper]) - - private def serialize(input: ExecutorStageSummaryWrapper): Array[Byte] = { + override def serialize(input: ExecutorStageSummaryWrapper): Array[Byte] = { val info = ExecutorStageSummarySerializer.serialize(input.info) val builder = StoreTypes.ExecutorStageSummaryWrapper.newBuilder() .setStageId(input.stageId.toLong) diff --git a/core/src/main/scala/org/apache/spark/status/protobuf/ExecutorSummaryWrapperSerializer.scala b/core/src/main/scala/org/apache/spark/status/protobuf/ExecutorSummaryWrapperSerializer.scala index 03a810157d7..b008c98e562 100644 --- a/core/src/main/scala/org/apache/spark/status/protobuf/ExecutorSummaryWrapperSerializer.scala +++ b/core/src/main/scala/org/apache/spark/status/protobuf/ExecutorSummaryWrapperSerializer.scala @@ -26,15 +26,9 @@ import org.apache.spark.status.ExecutorSummaryWrapper import org.apache.spark.status.api.v1.{ExecutorSummary, MemoryMetrics} import org.apache.spark.status.protobuf.Utils.getOptional -class ExecutorSummaryWrapperSerializer extends ProtobufSerDe { +class ExecutorSummaryWrapperSerializer extends ProtobufSerDe[ExecutorSummaryWrapper] { - override val supportClass: Class[_] = classOf[ExecutorSummaryWrapper] - - override def serialize(input: Any): Array[Byte] = { - serialize(input.asInstanceOf[ExecutorSummaryWrapper]) - } - - def serialize(input: ExecutorSummaryWrapper): Array[Byte] = { + override def serialize(input: ExecutorSummaryWrapper): Array[Byte] = { val info = serializeExecutorSummary(input.info) val builder = StoreTypes.ExecutorSummaryWrapper.newBuilder() .setInfo(info) diff --git a/core/src/main/scala/org/apache/spark/status/protobuf/JobDataWrapperSerializer.scala b/core/src/main/scala/org/apache/spark/status/protobuf/JobDataWrapperSerializer.scala index e2e2a1a8d89..10e0f125f6c 100644 --- a/core/src/main/scala/org/apache/spark/status/protobuf/JobDataWrapperSerializer.scala +++ b/core/src/main/scala/org/apache/spark/status/protobuf/JobDataWrapperSerializer.scala @@ -25,14 +25,9 @@ import org.apache.spark.status.JobDataWrapper import org.apache.spark.status.api.v1.JobData import org.apache.spark.status.protobuf.Utils.getOptional -class JobDataWrapperSerializer extends ProtobufSerDe { +class JobDataWrapperSerializer extends ProtobufSerDe[JobDataWrapper] { - override val supportClass: Class[_] = classOf[JobDataWrapper] - - override def serialize(input: Any): Array[Byte] = - serialize(input.asInstanceOf[JobDataWrapper]) - - private def serialize(j: JobDataWrapper): Array[Byte] = { + override def serialize(j: JobDataWrapper): Array[Byte] = { val jobData = serializeJobData(j.info) val builder = StoreTypes.JobDataWrapper.newBuilder() builder.setInfo(jobData) 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 d915edf396c..e6bdfa17715 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,6 +17,7 @@ package org.apache.spark.status.protobuf +import java.lang.reflect.ParameterizedType import java.util.ServiceLoader import collection.JavaConverters._ @@ -40,10 +41,16 @@ private[spark] class KVStoreProtobufSerializer extends KVStoreScalaSerializer { private[spark] object KVStoreProtobufSerializer { - private[this] lazy val serializerMap: Map[Class[_], ProtobufSerDe] = - ServiceLoader.load(classOf[ProtobufSerDe]) - .asScala.map(serDe => serDe.supportClass -> serDe).toMap + private[this] lazy val serializerMap: Map[Class[_], ProtobufSerDe[Any]] = { + def getGenericsType(klass: Class[_]): Class[_] = { + klass.getGenericInterfaces.head.asInstanceOf[ParameterizedType] + .getActualTypeArguments.head.asInstanceOf[Class[_]] + } + ServiceLoader.load(classOf[ProtobufSerDe[Any]]).asScala.map { serDe => + getGenericsType(serDe.getClass) -> serDe + }.toMap + } - def getSerializer(klass: Class[_]): Option[ProtobufSerDe] = + def getSerializer(klass: Class[_]): Option[ProtobufSerDe[Any]] = serializerMap.get(klass) } diff --git a/core/src/main/scala/org/apache/spark/status/protobuf/ProcessSummaryWrapperSerializer.scala b/core/src/main/scala/org/apache/spark/status/protobuf/ProcessSummaryWrapperSerializer.scala index 6d8a8f863ad..a3d13ddd31f 100644 --- a/core/src/main/scala/org/apache/spark/status/protobuf/ProcessSummaryWrapperSerializer.scala +++ b/core/src/main/scala/org/apache/spark/status/protobuf/ProcessSummaryWrapperSerializer.scala @@ -25,15 +25,9 @@ import org.apache.spark.status.ProcessSummaryWrapper import org.apache.spark.status.api.v1.ProcessSummary import org.apache.spark.status.protobuf.Utils.getOptional -class ProcessSummaryWrapperSerializer extends ProtobufSerDe { +class ProcessSummaryWrapperSerializer extends ProtobufSerDe[ProcessSummaryWrapper] { - override val supportClass: Class[_] = classOf[ProcessSummaryWrapper] - - override def serialize(input: Any): Array[Byte] = { - serialize(input.asInstanceOf[ProcessSummaryWrapper]) - } - - def serialize(input: ProcessSummaryWrapper): Array[Byte] = { + override def serialize(input: ProcessSummaryWrapper): Array[Byte] = { val builder = StoreTypes.ProcessSummaryWrapper.newBuilder() builder.setInfo(serializeProcessSummary(input.info)) builder.build().toByteArray diff --git a/core/src/main/scala/org/apache/spark/status/protobuf/ProtobufSerDe.scala b/core/src/main/scala/org/apache/spark/status/protobuf/ProtobufSerDe.scala index 5e0f6263f1a..d6eccb6307a 100644 --- a/core/src/main/scala/org/apache/spark/status/protobuf/ProtobufSerDe.scala +++ b/core/src/main/scala/org/apache/spark/status/protobuf/ProtobufSerDe.scala @@ -26,29 +26,20 @@ import org.apache.spark.annotation.{DeveloperApi, Unstable} * register itself to `org.apache.spark.status.protobuf.ProtobufSerDe` so that * `KVStoreProtobufSerializer` can use `ServiceLoader` to load and use them. * - * TODO: SPARK-41644 How to define `ProtobufSerDe` as `ProtobufSerDe[T]` - * * @since 3.4.0 */ @DeveloperApi @Unstable -trait ProtobufSerDe { - - /** - * Specify the data types supported by the current `ProtobufSerDe` - */ - val supportClass: Class[_] +trait ProtobufSerDe[T] { /** - * Serialize the input data of the type corresponding to `supportClass` - * to `Array[Byte]`, since the current input parameter type is `Any`, - * the input type needs to be guaranteed from the code level. + * Serialize the input data of the type `T` to `Array[Byte]`. */ - def serialize(input: Any): Array[Byte] + def serialize(input: T): Array[Byte] /** * Deserialize the input `Array[Byte]` to an object of the - * type corresponding to `supportClass`. + * type `T`. */ - def deserialize(bytes: Array[Byte]): Any + def deserialize(bytes: Array[Byte]): T } diff --git a/core/src/main/scala/org/apache/spark/status/protobuf/RDDOperationGraphWrapperSerializer.scala b/core/src/main/scala/org/apache/spark/status/protobuf/RDDOperationGraphWrapperSerializer.scala index c0d86ede198..f822ed1889a 100644 --- a/core/src/main/scala/org/apache/spark/status/protobuf/RDDOperationGraphWrapperSerializer.scala +++ b/core/src/main/scala/org/apache/spark/status/protobuf/RDDOperationGraphWrapperSerializer.scala @@ -24,12 +24,9 @@ import org.apache.spark.status.{RDDOperationClusterWrapper, RDDOperationGraphWra import org.apache.spark.status.protobuf.StoreTypes.{DeterministicLevel => GDeterministicLevel} import org.apache.spark.ui.scope.{RDDOperationEdge, RDDOperationNode} -class RDDOperationGraphWrapperSerializer extends ProtobufSerDe { +class RDDOperationGraphWrapperSerializer extends ProtobufSerDe[RDDOperationGraphWrapper] { - override val supportClass: Class[_] = classOf[RDDOperationGraphWrapper] - - override def serialize(input: Any): Array[Byte] = { - val op = input.asInstanceOf[RDDOperationGraphWrapper] + override def serialize(op: RDDOperationGraphWrapper): Array[Byte] = { val builder = StoreTypes.RDDOperationGraphWrapper.newBuilder() builder.setStageId(op.stageId.toLong) op.edges.foreach { e => diff --git a/core/src/main/scala/org/apache/spark/status/protobuf/RDDStorageInfoWrapperSerializer.scala b/core/src/main/scala/org/apache/spark/status/protobuf/RDDStorageInfoWrapperSerializer.scala index be6fe1f83cd..e59d363243e 100644 --- a/core/src/main/scala/org/apache/spark/status/protobuf/RDDStorageInfoWrapperSerializer.scala +++ b/core/src/main/scala/org/apache/spark/status/protobuf/RDDStorageInfoWrapperSerializer.scala @@ -23,14 +23,9 @@ import org.apache.spark.status.RDDStorageInfoWrapper import org.apache.spark.status.api.v1.{RDDDataDistribution, RDDPartitionInfo, RDDStorageInfo} import org.apache.spark.status.protobuf.Utils.getOptional -class RDDStorageInfoWrapperSerializer extends ProtobufSerDe { +class RDDStorageInfoWrapperSerializer extends ProtobufSerDe[RDDStorageInfoWrapper] { - override val supportClass: Class[_] = classOf[RDDStorageInfoWrapper] - - override def serialize(input: Any): Array[Byte] = - serialize(input.asInstanceOf[RDDStorageInfoWrapper]) - - private def serialize(input: RDDStorageInfoWrapper): Array[Byte] = { + override def serialize(input: RDDStorageInfoWrapper): Array[Byte] = { val builder = StoreTypes.RDDStorageInfoWrapper.newBuilder() builder.setInfo(serializeRDDStorageInfo(input.info)) builder.build().toByteArray diff --git a/core/src/main/scala/org/apache/spark/status/protobuf/ResourceProfileWrapperSerializer.scala b/core/src/main/scala/org/apache/spark/status/protobuf/ResourceProfileWrapperSerializer.scala index a780ea5d7e9..d9d29cc8d88 100644 --- a/core/src/main/scala/org/apache/spark/status/protobuf/ResourceProfileWrapperSerializer.scala +++ b/core/src/main/scala/org/apache/spark/status/protobuf/ResourceProfileWrapperSerializer.scala @@ -19,16 +19,11 @@ package org.apache.spark.status.protobuf import org.apache.spark.status.ResourceProfileWrapper -class ResourceProfileWrapperSerializer extends ProtobufSerDe { +class ResourceProfileWrapperSerializer extends ProtobufSerDe[ResourceProfileWrapper] { private val appEnvSerializer = new ApplicationEnvironmentInfoWrapperSerializer - override val supportClass: Class[_] = classOf[ResourceProfileWrapper] - - override def serialize(input: Any): Array[Byte] = - serialize(input.asInstanceOf[ResourceProfileWrapper]) - - private def serialize(input: ResourceProfileWrapper): Array[Byte] = { + override def serialize(input: ResourceProfileWrapper): Array[Byte] = { val builder = StoreTypes.ResourceProfileWrapper.newBuilder() builder.setRpInfo(appEnvSerializer.serializeResourceProfileInfo(input.rpInfo)) builder.build().toByteArray 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 index 936e2dd4504..1b9a1ecfce6 100644 --- a/core/src/main/scala/org/apache/spark/status/protobuf/SpeculationStageSummaryWrapperSerializer.scala +++ b/core/src/main/scala/org/apache/spark/status/protobuf/SpeculationStageSummaryWrapperSerializer.scala @@ -20,14 +20,10 @@ package org.apache.spark.status.protobuf import org.apache.spark.status.SpeculationStageSummaryWrapper import org.apache.spark.status.api.v1.SpeculationStageSummary -class SpeculationStageSummaryWrapperSerializer extends ProtobufSerDe { +class SpeculationStageSummaryWrapperSerializer + extends ProtobufSerDe[SpeculationStageSummaryWrapper] { - override val supportClass: Class[_] = classOf[SpeculationStageSummaryWrapper] - - override def serialize(input: Any): Array[Byte] = - serialize(input.asInstanceOf[SpeculationStageSummaryWrapper]) - - private def serialize(s: SpeculationStageSummaryWrapper): Array[Byte] = { + override def serialize(s: SpeculationStageSummaryWrapper): Array[Byte] = { val summary = serializeSpeculationStageSummary(s.info) val builder = StoreTypes.SpeculationStageSummaryWrapper.newBuilder() builder.setStageId(s.stageId.toLong) diff --git a/core/src/main/scala/org/apache/spark/status/protobuf/StageDataWrapperSerializer.scala b/core/src/main/scala/org/apache/spark/status/protobuf/StageDataWrapperSerializer.scala index f4e5c8104a0..eda4422405e 100644 --- a/core/src/main/scala/org/apache/spark/status/protobuf/StageDataWrapperSerializer.scala +++ b/core/src/main/scala/org/apache/spark/status/protobuf/StageDataWrapperSerializer.scala @@ -27,16 +27,13 @@ import org.apache.spark.status.api.v1.{ExecutorMetricsDistributions, ExecutorPea import org.apache.spark.status.protobuf.Utils.getOptional import org.apache.spark.util.Utils.weakIntern -class StageDataWrapperSerializer extends ProtobufSerDe { +class StageDataWrapperSerializer extends ProtobufSerDe[StageDataWrapper] { - override val supportClass: Class[_] = classOf[StageDataWrapper] - - override def serialize(input: Any): Array[Byte] = { - val s = input.asInstanceOf[StageDataWrapper] + override def serialize(input: StageDataWrapper): Array[Byte] = { val builder = StoreTypes.StageDataWrapper.newBuilder() - builder.setInfo(serializeStageData(s.info)) - s.jobIds.foreach(id => builder.addJobIds(id.toLong)) - s.locality.foreach { entry => + builder.setInfo(serializeStageData(input.info)) + input.jobIds.foreach(id => builder.addJobIds(id.toLong)) + input.locality.foreach { entry => builder.putLocality(entry._1, entry._2) } builder.build().toByteArray diff --git a/core/src/main/scala/org/apache/spark/status/protobuf/StreamBlockDataSerializer.scala b/core/src/main/scala/org/apache/spark/status/protobuf/StreamBlockDataSerializer.scala index 71bb09d3118..f450bbbfd0c 100644 --- a/core/src/main/scala/org/apache/spark/status/protobuf/StreamBlockDataSerializer.scala +++ b/core/src/main/scala/org/apache/spark/status/protobuf/StreamBlockDataSerializer.scala @@ -19,11 +19,9 @@ package org.apache.spark.status.protobuf import org.apache.spark.status.StreamBlockData -class StreamBlockDataSerializer extends ProtobufSerDe { - override val supportClass: Class[_] = classOf[StreamBlockData] +class StreamBlockDataSerializer extends ProtobufSerDe[StreamBlockData] { - override def serialize(input: Any): Array[Byte] = { - val data = input.asInstanceOf[StreamBlockData] + override def serialize(data: StreamBlockData): Array[Byte] = { val builder = StoreTypes.StreamBlockData.newBuilder() .setName(data.name) .setExecutorId(data.executorId) diff --git a/core/src/main/scala/org/apache/spark/status/protobuf/TaskDataWrapperSerializer.scala b/core/src/main/scala/org/apache/spark/status/protobuf/TaskDataWrapperSerializer.scala index 7ba4a32d806..3c947c79eab 100644 --- a/core/src/main/scala/org/apache/spark/status/protobuf/TaskDataWrapperSerializer.scala +++ b/core/src/main/scala/org/apache/spark/status/protobuf/TaskDataWrapperSerializer.scala @@ -21,14 +21,9 @@ import org.apache.spark.status.TaskDataWrapper import org.apache.spark.status.protobuf.Utils.getOptional import org.apache.spark.util.Utils.weakIntern -class TaskDataWrapperSerializer extends ProtobufSerDe { +class TaskDataWrapperSerializer extends ProtobufSerDe[TaskDataWrapper] { - override val supportClass: Class[_] = classOf[TaskDataWrapper] - - override def serialize(input: Any): Array[Byte] = - serialize(input.asInstanceOf[TaskDataWrapper]) - - private def serialize(input: TaskDataWrapper): Array[Byte] = { + override def serialize(input: TaskDataWrapper): Array[Byte] = { val builder = StoreTypes.TaskDataWrapper.newBuilder() .setTaskId(input.taskId) .setIndex(input.index) diff --git a/sql/core/src/main/scala/org/apache/spark/status/protobuf/sql/SQLExecutionUIDataSerializer.scala b/sql/core/src/main/scala/org/apache/spark/status/protobuf/sql/SQLExecutionUIDataSerializer.scala index dcce69b803c..80a36e1b02b 100644 --- a/sql/core/src/main/scala/org/apache/spark/status/protobuf/sql/SQLExecutionUIDataSerializer.scala +++ b/sql/core/src/main/scala/org/apache/spark/status/protobuf/sql/SQLExecutionUIDataSerializer.scala @@ -25,12 +25,9 @@ import org.apache.spark.sql.execution.ui.SQLExecutionUIData import org.apache.spark.status.protobuf.{JobExecutionStatusSerializer, ProtobufSerDe, StoreTypes} import org.apache.spark.status.protobuf.Utils.getOptional -class SQLExecutionUIDataSerializer extends ProtobufSerDe { +class SQLExecutionUIDataSerializer extends ProtobufSerDe[SQLExecutionUIData] { - override val supportClass: Class[_] = classOf[SQLExecutionUIData] - - override def serialize(input: Any): Array[Byte] = { - val ui = input.asInstanceOf[SQLExecutionUIData] + override def serialize(ui: SQLExecutionUIData): Array[Byte] = { val builder = StoreTypes.SQLExecutionUIData.newBuilder() builder.setExecutionId(ui.executionId) builder.setRootExecutionId(ui.rootExecutionId) diff --git a/sql/core/src/main/scala/org/apache/spark/status/protobuf/sql/SparkPlanGraphWrapperSerializer.scala b/sql/core/src/main/scala/org/apache/spark/status/protobuf/sql/SparkPlanGraphWrapperSerializer.scala index c68466489ce..1df82e3246a 100644 --- a/sql/core/src/main/scala/org/apache/spark/status/protobuf/sql/SparkPlanGraphWrapperSerializer.scala +++ b/sql/core/src/main/scala/org/apache/spark/status/protobuf/sql/SparkPlanGraphWrapperSerializer.scala @@ -23,12 +23,9 @@ import org.apache.spark.sql.execution.ui.{SparkPlanGraphClusterWrapper, SparkPla import org.apache.spark.status.protobuf.ProtobufSerDe import org.apache.spark.status.protobuf.StoreTypes -class SparkPlanGraphWrapperSerializer extends ProtobufSerDe { +class SparkPlanGraphWrapperSerializer extends ProtobufSerDe[SparkPlanGraphWrapper] { - override val supportClass: Class[_] = classOf[SparkPlanGraphWrapper] - - override def serialize(input: Any): Array[Byte] = { - val plan = input.asInstanceOf[SparkPlanGraphWrapper] + override def serialize(plan: SparkPlanGraphWrapper): Array[Byte] = { val builder = StoreTypes.SparkPlanGraphWrapper.newBuilder() builder.setExecutionId(plan.executionId) plan.nodes.foreach { node => diff --git a/sql/core/src/main/scala/org/apache/spark/status/protobuf/sql/StreamingQueryDataSerializer.scala b/sql/core/src/main/scala/org/apache/spark/status/protobuf/sql/StreamingQueryDataSerializer.scala index f05b186fea5..70f8bedf91b 100644 --- a/sql/core/src/main/scala/org/apache/spark/status/protobuf/sql/StreamingQueryDataSerializer.scala +++ b/sql/core/src/main/scala/org/apache/spark/status/protobuf/sql/StreamingQueryDataSerializer.scala @@ -23,12 +23,9 @@ import org.apache.spark.sql.streaming.ui.StreamingQueryData import org.apache.spark.status.protobuf.{ProtobufSerDe, StoreTypes} import org.apache.spark.status.protobuf.Utils.getOptional -class StreamingQueryDataSerializer extends ProtobufSerDe { +class StreamingQueryDataSerializer extends ProtobufSerDe[StreamingQueryData] { - override val supportClass: Class[_] = classOf[StreamingQueryData] - - override def serialize(input: Any): Array[Byte] = { - val data = input.asInstanceOf[StreamingQueryData] + override def serialize(data: StreamingQueryData): Array[Byte] = { val builder = StoreTypes.StreamingQueryData.newBuilder() .setId(data.id.toString) .setRunId(data.runId) @@ -40,7 +37,7 @@ class StreamingQueryDataSerializer extends ProtobufSerDe { builder.build().toByteArray } - override def deserialize(bytes: Array[Byte]): Any = { + override def deserialize(bytes: Array[Byte]): StreamingQueryData = { val data = StoreTypes.StreamingQueryData.parseFrom(bytes) val exception = getOptional(data.hasException, () => data.getException) --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org