[SPARK-41244][UI] Introducing a Protobuf serializer for UI data on KV store #38779
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -532,7 +532,12 @@ | |
| <groupId>org.apache.commons</groupId> | ||
| <artifactId>commons-crypto</artifactId> | ||
| </dependency> | ||
| <dependency> | ||
| <groupId>com.google.protobuf</groupId> | ||
| <artifactId>protobuf-java</artifactId> | ||
| <version>${protobuf.version}</version> | ||
| <scope>compile</scope> | ||
| </dependency> | ||
| </dependencies> | ||
| <build> | ||
| <outputDirectory>target/scala-${scala.binary.version}/classes</outputDirectory> | ||
|
|
@@ -616,6 +621,50 @@ | |
| </execution> | ||
| </executions> | ||
| </plugin> | ||
| <plugin> | ||
| <groupId>org.apache.maven.plugins</groupId> | ||
|
Contributor
There was a problem hiding this comment. SPARK-40593 and SPARK-41215 do some work to support Spark 3.4 to be compiled on CentOS6&7. I think the core module needs similar work, but may not be in this pr
Contributor
There was a problem hiding this comment. @WolverineJiang Are you interested in solving this problem after PR merge? Just like what you did in SPARK-41215 There was a problem hiding this comment.
More than willing, I'll start in a few days. |
||
| <artifactId>maven-shade-plugin</artifactId> | ||
|
Contributor
There was a problem hiding this comment. @gengliangwang It actually conflicts with parent pom.xml which causes the jetty can not be shaded into spark-core. The current master branch can not start Spark UI, see error message:
Member
Author
There was a problem hiding this comment. @ulysses-you Thanks for reporting it. Do you know how to fix it?
Contributor
There was a problem hiding this comment. maybe should add
Member
There was a problem hiding this comment. Ya, I also saw this when I build distributions. Thank you, @pan3793 ! |
||
| <configuration> | ||
| <shadedArtifactAttached>false</shadedArtifactAttached> | ||
| <shadeTestJar>true</shadeTestJar> | ||
| <artifactSet> | ||
| <includes> | ||
| <include>com.google.protobuf:*</include> | ||
| </includes> | ||
| </artifactSet> | ||
| <relocations> | ||
| <relocation> | ||
| <pattern>com.google.protobuf</pattern> | ||
| <shadedPattern>${spark.shade.packageName}.spark-core.protobuf</shadedPattern> | ||
| <includes> | ||
| <include>com.google.protobuf.**</include> | ||
| </includes> | ||
| </relocation> | ||
| </relocations> | ||
| </configuration> | ||
| </plugin> | ||
| <plugin> | ||
| <groupId>com.github.os72</groupId> | ||
| <artifactId>protoc-jar-maven-plugin</artifactId> | ||
| <version>3.11.4</version> | ||
cloud-fan marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| <!-- Generates Java classes for tests. TODO(Raghu): Generate descriptor files too. --> | ||
| <executions> | ||
| <execution> | ||
| <phase>generate-test-sources</phase> | ||
gengliangwang marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| <goals> | ||
| <goal>run</goal> | ||
| </goals> | ||
| <configuration> | ||
| <protocArtifact>com.google.protobuf:protoc:${protobuf.version}</protocArtifact> | ||
| <protocVersion>${protobuf.version}</protocVersion> | ||
| <inputDirectories> | ||
| <include>src/main/protobuf</include> | ||
| </inputDirectories> | ||
| <addSources>test</addSources> | ||
gengliangwang marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| </configuration> | ||
| </execution> | ||
| </executions> | ||
| </plugin> | ||
| </plugins> | ||
| </build> | ||
|
|
||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| /* | ||
| * 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. | ||
| */ | ||
|
|
||
| syntax = "proto3"; | ||
| package org.apache.spark.status.protobuf; | ||
|
|
||
| message JavaDate { | ||
| // The number of milliseconds since January 1, 1970, 00:00:00 GMT | ||
| optional int64 date = 1; | ||
|
||
| } | ||
|
|
||
| enum JobExecutionStatus { | ||
| RUNNING = 0; | ||
|
||
| SUCCEEDED = 1; | ||
| FAILED = 2; | ||
| UNKNOWN = 3; | ||
| } | ||
|
|
||
| message JobData { | ||
| int32 job_id = 1; | ||
|
||
| string name = 2; | ||
| optional string description = 3; | ||
| optional JavaDate submission_time = 4; | ||
| optional JavaDate completion_time = 5; | ||
mridulm marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| repeated int32 stage_ids = 6; | ||
| optional string job_group = 7; | ||
| JobExecutionStatus status = 8; | ||
| int32 num_tasks = 9; | ||
| int32 num_active_tasks = 10; | ||
| int32 num_completed_tasks = 11; | ||
| int32 num_skipped_tasks = 12; | ||
| int32 num_failed_tasks = 13; | ||
| int32 num_killed_tasks = 14; | ||
| int32 num_completed_indices = 15; | ||
| int32 num_active_stages = 16; | ||
| int32 num_completed_stages = 17; | ||
| int32 num_skipped_stages = 18; | ||
| int32 num_failed_stages = 19; | ||
| map<string, int32> kill_tasks_summary = 20; | ||
| } | ||
|
|
||
| message JobDataWrapper { | ||
| JobData info = 1; | ||
| repeated int32 skipped_stages = 2; | ||
| optional int64 sql_execution_id = 3; | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,128 @@ | ||
| /* | ||
| * 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 java.util.Date | ||
|
|
||
| import org.apache.spark.JobExecutionStatus | ||
| import org.apache.spark.status.JobDataWrapper | ||
| import org.apache.spark.status.api.v1.JobData | ||
|
|
||
| object JobDataWrapperSerializer { | ||
| def serialize(j: JobDataWrapper): Array[Byte] = { | ||
| val jobData = serializeJobData(j.info) | ||
| val builder = StoreTypes.JobDataWrapper.newBuilder() | ||
|
Contributor
There was a problem hiding this comment. Where can we find
Member
Author
There was a problem hiding this comment. It is generated during compile time. It should be under |
||
| builder.setInfo(jobData) | ||
| j.skippedStages.foreach(builder.addSkippedStages) | ||
| j.sqlExecutionId.foreach(builder.setSqlExecutionId) | ||
| builder.build().toByteArray | ||
| } | ||
|
|
||
| def deserialize(bytes: Array[Byte]): JobDataWrapper = { | ||
| val wrapper = StoreTypes.JobDataWrapper.parseFrom(bytes) | ||
| val sqlExecutionId = getOptional(wrapper.hasSqlExecutionId, wrapper.getSqlExecutionId) | ||
| new JobDataWrapper( | ||
| deserializeJobData(wrapper.getInfo), | ||
| wrapper.getSkippedStagesList.asScala.map(_.toInt).toSet, | ||
| sqlExecutionId | ||
| ) | ||
| } | ||
|
|
||
| private def getOptional[T](condition: Boolean, result: () => T): Option[T] = if (condition) { | ||
| Some(result()) | ||
| } else { | ||
| None | ||
| } | ||
|
|
||
| private def serializeJobData(jobData: JobData): StoreTypes.JobData = { | ||
| val jobDataBuilder = StoreTypes.JobData.newBuilder() | ||
| jobDataBuilder.setJobId(jobData.jobId) | ||
| .setName(jobData.name) | ||
| .setStatus(serializeJobExecutionStatus(jobData.status)) | ||
| .setNumTasks(jobData.numTasks) | ||
| .setNumActiveTasks(jobData.numActiveTasks) | ||
| .setNumCompletedTasks(jobData.numCompletedTasks) | ||
| .setNumSkippedTasks(jobData.numSkippedTasks) | ||
| .setNumFailedTasks(jobData.numFailedTasks) | ||
| .setNumKilledTasks(jobData.numKilledTasks) | ||
| .setNumCompletedIndices(jobData.numCompletedIndices) | ||
| .setNumActiveStages(jobData.numActiveStages) | ||
| .setNumCompletedStages(jobData.numCompletedStages) | ||
| .setNumSkippedStages(jobData.numSkippedStages) | ||
| .setNumFailedStages(jobData.numFailedStages) | ||
|
|
||
| jobData.description.foreach(jobDataBuilder.setDescription) | ||
| jobData.submissionTime.foreach { d => | ||
| jobDataBuilder.setSubmissionTime(serializeDate(d)) | ||
| } | ||
| jobData.completionTime.foreach { d => | ||
| jobDataBuilder.setCompletionTime(serializeDate(d)) | ||
| } | ||
| jobData.stageIds.foreach(jobDataBuilder.addStageIds) | ||
| jobData.jobGroup.foreach(jobDataBuilder.setJobGroup) | ||
| jobData.killedTasksSummary.foreach { entry => | ||
| jobDataBuilder.putKillTasksSummary(entry._1, entry._2) | ||
| } | ||
| jobDataBuilder.build() | ||
| } | ||
|
|
||
| private def deserializeJobData(info: StoreTypes.JobData): JobData = { | ||
| val description = if (info.hasDescription) { | ||
gengliangwang marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| Some(info.getDescription) | ||
| } else { | ||
| None | ||
| } | ||
| val submissionTime = | ||
| getOptional(info.hasSubmissionTime, () => new Date(info.getSubmissionTime.getDate)) | ||
| val completionTime = getOptional(info.hasCompletionTime, | ||
| () => new Date(info.getCompletionTime.getDate)) | ||
| val jobGroup = getOptional(info.hasJobGroup, info.getJobGroup) | ||
| val status = JobExecutionStatus.valueOf(info.getStatus.toString) | ||
|
|
||
| new JobData( | ||
| jobId = info.getJobId, | ||
| name = info.getName, | ||
| description = description, | ||
| submissionTime = submissionTime, | ||
| completionTime = completionTime, | ||
| stageIds = info.getStageIdsList.asScala.map(_.toInt), | ||
|
||
| jobGroup = jobGroup, | ||
| status = status, | ||
| numTasks = info.getNumTasks, | ||
| numActiveTasks = info.getNumActiveTasks, | ||
| numCompletedTasks = info.getNumCompletedTasks, | ||
| numSkippedTasks = info.getNumSkippedTasks, | ||
| numFailedTasks = info.getNumFailedTasks, | ||
| numKilledTasks = info.getNumKilledTasks, | ||
| numCompletedIndices = info.getNumCompletedIndices, | ||
| numActiveStages = info.getNumActiveStages, | ||
| numCompletedStages = info.getNumCompletedStages, | ||
| numSkippedStages = info.getNumSkippedStages, | ||
| numFailedStages = info.getNumFailedStages, | ||
| killedTasksSummary = info.getKillTasksSummaryMap.asScala.mapValues(_.toInt).toMap) | ||
| } | ||
|
|
||
| private def serializeDate(d: Date): StoreTypes.JavaDate = { | ||
| StoreTypes.JavaDate.newBuilder().setDate(d.getTime).build() | ||
| } | ||
|
|
||
| private def serializeJobExecutionStatus(j: JobExecutionStatus): StoreTypes.JobExecutionStatus = { | ||
| StoreTypes.JobExecutionStatus.valueOf(j.toString) | ||
| } | ||
| } | ||
Uh oh!
There was an error while loading. Please reload this page.