-
Notifications
You must be signed in to change notification settings - Fork 1.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Extract all yarn related methods to azkaban-common as shared util lib…
…rary
- Loading branch information
1 parent
f7ad185
commit a35ff0e
Showing
5 changed files
with
150 additions
and
73 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| package azkaban.utils; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Set; | ||
| import org.apache.hadoop.fs.Path; | ||
| import org.apache.log4j.Logger; | ||
| import org.apache.hadoop.yarn.api.records.ApplicationId; | ||
| import org.apache.hadoop.yarn.client.api.YarnClient; | ||
| import org.apache.hadoop.yarn.conf.YarnConfiguration; | ||
| import org.apache.hadoop.yarn.exceptions.YarnException; | ||
|
|
||
| public class YarnUtils { | ||
|
|
||
| //Yarn resource configuration directory for the cluster where the job is scheduled by the cluster router | ||
| private static final String YARN_CONF_DIRECTORY_PROPERTY = "env.YARN_CONF_DIR"; | ||
| private static final String YARN_CONF_FILENAME = "yarn-site.xml"; | ||
|
|
||
| /** | ||
| * Uses YarnClient to kill the jobs one by one | ||
| */ | ||
| public static void killAllAppsOnCluster(YarnClient yarnClient, Set<String> applicationIDs, | ||
| Logger log) { | ||
| log.info(String.format("Killing applications: %s", applicationIDs)); | ||
|
|
||
| for (final String appId : applicationIDs) { | ||
| try { | ||
| YarnUtils.killAppOnCluster(yarnClient, appId, log); | ||
| } catch (final Throwable t) { | ||
| log.warn("something happened while trying to kill this job: " + appId, t); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * <pre> | ||
| * Uses YarnClient to kill the job on the Hadoop Yarn Cluster. | ||
| * Using JobClient only works partially: | ||
| * If yarn container has started but spark job haven't, it will kill | ||
| * If spark job has started, the cancel will hang until the spark job is complete | ||
| * If the spark job is complete, it will return immediately, with a job not found on job tracker | ||
| * </pre> | ||
| */ | ||
| public static void killAppOnCluster(final YarnClient yarnClient, final String applicationId, | ||
| final Logger log) throws YarnException, IOException { | ||
|
|
||
| final String[] split = applicationId.split("_"); | ||
| final ApplicationId aid = ApplicationId.newInstance(Long.parseLong(split[1]), | ||
| Integer.parseInt(split[2])); | ||
| log.info("start killing application: " + aid); | ||
| yarnClient.killApplication(aid); | ||
| log.info("successfully killed application: " + aid); | ||
| } | ||
|
|
||
| /** | ||
| * Create, initialize and start a YarnClient connecting to the Yarn Cluster (resource manager), | ||
| * using the resources passed in with props. | ||
| * | ||
| * @param props the properties to create a YarnClient, the path to the "yarn-site.xml" to be used | ||
| */ | ||
| public static YarnClient createYarnClient(Props props) { | ||
| final YarnConfiguration yarnConf = new YarnConfiguration(); | ||
| if (props.containsKey(YARN_CONF_DIRECTORY_PROPERTY)) { | ||
| yarnConf.addResource( | ||
| new Path(props.get(YARN_CONF_DIRECTORY_PROPERTY) + "/" + YARN_CONF_FILENAME)); | ||
| } | ||
| final YarnClient yarnClient = YarnClient.createYarnClient(); | ||
| yarnClient.init(yarnConf); | ||
| yarnClient.start(); | ||
| return yarnClient; | ||
| } | ||
| } |
63 changes: 63 additions & 0 deletions
63
azkaban-common/src/test/java/azkaban/utils/YarnUtilsTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| package azkaban.utils; | ||
|
|
||
|
|
||
| import static org.mockito.ArgumentMatchers.any; | ||
| import static org.mockito.ArgumentMatchers.eq; | ||
| import static org.mockito.Mockito.doNothing; | ||
| import static org.mockito.Mockito.doThrow; | ||
| import static org.mockito.Mockito.mock; | ||
| import static org.mockito.Mockito.when; | ||
|
|
||
| import com.google.common.collect.ImmutableSet; | ||
| import java.io.IOException; | ||
| import org.apache.hadoop.yarn.api.records.ApplicationId; | ||
| import org.apache.hadoop.yarn.client.api.YarnClient; | ||
| import org.apache.hadoop.yarn.exceptions.YarnException; | ||
| import org.apache.log4j.Logger; | ||
| import org.junit.Test; | ||
|
|
||
| public class YarnUtilsTest { | ||
|
|
||
| final private Logger log = Logger.getLogger(YarnUtilsTest.class); | ||
|
|
||
| @Test | ||
| public void testKillAppOnCluster() throws IOException, YarnException { | ||
| YarnClient mockClient = mock(YarnClient.class); | ||
| doNothing().when(mockClient).killApplication(any()); | ||
| YarnUtils.killAppOnCluster(mockClient, "application_123_456", log); | ||
| } | ||
|
|
||
| @Test(expected = IndexOutOfBoundsException.class) | ||
| public void testKillAppOnClusterInvalidAppID() throws IOException, YarnException { | ||
| YarnClient mockClient = mock(YarnClient.class); | ||
| doNothing().when(mockClient).killApplication(any()); | ||
| YarnUtils.killAppOnCluster(mockClient, "application+++123===456", log); | ||
| } | ||
|
|
||
| @Test(expected = YarnException.class) | ||
| public void testKillAppOnClusterYarnFail() throws IOException, YarnException { | ||
| YarnClient mockClient = mock(YarnClient.class); | ||
| doThrow(new YarnException("ops")).when(mockClient).killApplication(any()); | ||
| YarnUtils.killAppOnCluster(mockClient, "application_123_456", log); | ||
| } | ||
|
|
||
| @Test | ||
| public void testKillAllAppsOnCluster() throws IOException, YarnException { | ||
| YarnClient mockClient = mock(YarnClient.class); | ||
| doNothing().when(mockClient).killApplication(any()); | ||
| YarnUtils.killAllAppsOnCluster(mockClient, | ||
| ImmutableSet.of("application_123_456", "application_456_123"), log); | ||
| } | ||
|
|
||
| @Test | ||
| public void testKillAllAppsOnClusterFail() throws IOException, YarnException { | ||
| YarnClient mockClient = mock(YarnClient.class); | ||
| doNothing().when(mockClient).killApplication(any()); | ||
| doThrow(new YarnException("ops")).when(mockClient).killApplication( | ||
| eq(ApplicationId.newInstance(4560, 1230))); | ||
|
|
||
| // log the error but not throw | ||
| YarnUtils.killAllAppsOnCluster(mockClient, | ||
| ImmutableSet.of("application_123_456", "application_4560_1230"), log); | ||
| } | ||
| } |