Skip to content

Commit b54a0c4

Browse files
author
Andrew Or
committed
Initial skeleton for Yarn shuffle service
This includes the necessary changes in making the Yarn shuffle service a module of Spark that core doesn't depend on. This is included in the network-yarn module, which depends on the network-shuffle module.
1 parent 2aca97c commit b54a0c4

File tree

4 files changed

+145
-3
lines changed

4 files changed

+145
-3
lines changed

‎network/yarn/pom.xml‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<!--
3+
~ Licensed to the Apache Software Foundation (ASF) under one or more
4+
~ contributor license agreements. See the NOTICE file distributed with
5+
~ this work for additional information regarding copyright ownership.
6+
~ The ASF licenses this file to You under the Apache License, Version 2.0
7+
~ (the "License"); you may not use this file except in compliance with
8+
~ the License. You may obtain a copy of the License at
9+
~
10+
~ http://www.apache.org/licenses/LICENSE-2.0
11+
~
12+
~ Unless required by applicable law or agreed to in writing, software
13+
~ distributed under the License is distributed on an "AS IS" BASIS,
14+
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
~ See the License for the specific language governing permissions and
16+
~ limitations under the License.
17+
-->
18+
19+
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
20+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
21+
<modelVersion>4.0.0</modelVersion>
22+
<parent>
23+
<groupId>org.apache.spark</groupId>
24+
<artifactId>spark-parent</artifactId>
25+
<version>1.2.0-SNAPSHOT</version>
26+
<relativePath>../../pom.xml</relativePath>
27+
</parent>
28+
29+
<groupId>org.apache.spark</groupId>
30+
<artifactId>spark-network-yarn_2.10</artifactId>
31+
<packaging>jar</packaging>
32+
<name>Spark Project Yarn Shuffle Service Code</name>
33+
<url>http://spark.apache.org/</url>
34+
<properties>
35+
<sbt.project.name>network-yarn</sbt.project.name>
36+
</properties>
37+
38+
<dependencies>
39+
<!-- Core dependencies -->
40+
<dependency>
41+
<groupId>org.apache.spark</groupId>
42+
<artifactId>spark-network-shuffle_2.10</artifactId>
43+
<version>${project.version}</version>
44+
</dependency>
45+
46+
<!-- Provided dependencies -->
47+
<dependency>
48+
<groupId>org.apache.hadoop</groupId>
49+
<artifactId>hadoop-yarn-api</artifactId>
50+
<scope>provided</scope>
51+
</dependency>
52+
<dependency>
53+
<groupId>org.apache.hadoop</groupId>
54+
<artifactId>hadoop-yarn-common</artifactId>
55+
<scope>provided</scope>
56+
</dependency>
57+
<dependency>
58+
<groupId>org.apache.hadoop</groupId>
59+
<artifactId>hadoop-yarn-server-web-proxy</artifactId>
60+
<scope>provided</scope>
61+
</dependency>
62+
<dependency>
63+
<groupId>org.apache.hadoop</groupId>
64+
<artifactId>hadoop-yarn-client</artifactId>
65+
<scope>provided</scope>
66+
</dependency>
67+
<dependency>
68+
<groupId>org.apache.hadoop</groupId>
69+
<artifactId>hadoop-client</artifactId>
70+
<scope>provided</scope>
71+
</dependency>
72+
</dependencies>
73+
74+
<build>
75+
<outputDirectory>target/scala-${scala.binary.version}/classes</outputDirectory>
76+
<testOutputDirectory>target/scala-${scala.binary.version}/test-classes</testOutputDirectory>
77+
</build>
78+
</project>

‎network/yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleService.java‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
1+
2+
package org.apache.spark.network.yarn;
3+
4+
import java.nio.ByteBuffer;
5+
import java.sql.Timestamp;
6+
import java.util.Date;
7+
8+
import org.apache.hadoop.mapreduce.security.token.JobTokenSecretManager;
9+
import org.apache.hadoop.yarn.api.records.ApplicationId;
10+
import org.apache.hadoop.yarn.api.records.ContainerId;
11+
import org.apache.hadoop.yarn.server.api.AuxiliaryService;
12+
import org.apache.hadoop.yarn.server.api.ApplicationInitializationContext;
13+
import org.apache.hadoop.yarn.server.api.ApplicationTerminationContext;
14+
import org.apache.hadoop.yarn.server.api.ContainerInitializationContext;
15+
import org.apache.hadoop.yarn.server.api.ContainerTerminationContext;
16+
17+
/**
18+
* External shuffle service used by Spark on Yarn.
19+
*/
20+
public class YarnShuffleService extends AuxiliaryService {
21+
22+
private static final JobTokenSecretManager secretManager = new JobTokenSecretManager();
23+
24+
public YarnShuffleService() {
25+
super("sparkshuffleservice");
26+
log("--- [ Welcome to YarnShuffleService v0.1 ] ---");
27+
}
28+
29+
@Override
30+
public void initializeApplication(ApplicationInitializationContext context) {
31+
ApplicationId appId = context.getApplicationId();
32+
log("Initializing application " + appId + "!");
33+
}
34+
35+
@Override
36+
public void stopApplication(ApplicationTerminationContext context) {
37+
ApplicationId appId = context.getApplicationId();
38+
log("Stopping application " + appId + "!");
39+
}
40+
41+
@Override
42+
public ByteBuffer getMetaData() {
43+
log("Getting meta data");
44+
return ByteBuffer.wrap("".getBytes());
45+
}
46+
47+
@Override
48+
public void initializeContainer(ContainerInitializationContext context) {
49+
ContainerId containerId = context.getContainerId();
50+
log("Initializing container " + containerId + "!");
51+
}
52+
53+
@Override
54+
public void stopContainer(ContainerTerminationContext context) {
55+
ContainerId containerId = context.getContainerId();
56+
log("Stopping container " + containerId + "!");
57+
}
58+
59+
private void log(String msg) {
60+
Timestamp timestamp = new Timestamp(new Date().getTime());
61+
System.out.println("* org.apache.spark.YarnShuffleService " + timestamp + ": " + msg);
62+
}
63+
}

‎pom.xml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,7 @@
9393
<module>tools</module>
9494
<module>network/common</module>
9595
<module>network/shuffle</module>
96+
<module>network/yarn</module>
9697
<module>streaming</module>
9798
<module>sql/catalyst</module>
9899
<module>sql/core</module>

‎project/SparkBuild.scala‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -42,8 +42,8 @@ object BuildCommons {
4242
Seq("yarn", "yarn-stable", "yarn-alpha", "java8-tests", "ganglia-lgpl", "kinesis-asl")
4343
.map(ProjectRef(buildLocation, _))
4444

45-
val assemblyProjects@Seq(assembly, examples) = Seq("assembly", "examples")
46-
.map(ProjectRef(buildLocation, _))
45+
val assemblyProjects@Seq(assembly, examples, networkYarn) =
46+
Seq("assembly", "examples", "network-yarn").map(ProjectRef(buildLocation, _))
4747

4848
val tools = ProjectRef(buildLocation, "tools")
4949
// Root project.
@@ -143,7 +143,7 @@ object SparkBuild extends PomBuild {
143143

144144
// TODO: Add Sql to mima checks
145145
allProjects.filterNot(x => Seq(spark, sql, hive, hiveThriftServer, catalyst, repl,
146-
streamingFlumeSink, networkCommon, networkShuffle).contains(x)).foreach {
146+
streamingFlumeSink, networkCommon, networkShuffle, networkYarn).contains(x)).foreach {
147147
x => enable(MimaBuild.mimaSettings(sparkHome, x))(x)
148148
}
149149

0 commit comments

Comments
 (0)