Yarn RM 写 ZNode 超数据量限制 bug 修复

关注

发布于: 2021 年 01 月 08 日

问题背景

线上集群出现过几次 Yarn RM 写 ZK ZNode 的数据量超过 ZNode 限制，导致 RM 服务均进入 Standby 状态，用户无法正常提交任务，整个集群 hang 住，后续排查发现主要是异常任务写 ZNode 数据量太大，超过 ZNode 限制，导致集群其他提交作业的状态信息无法正常写入 ZNode，为避免类似问题再次出现，我们对 RM 写 ZNode 逻辑进行了优化，规避异常任务对整个集群造成的雪崩效应。

一、问题复现

最直接方式是修改 ZK 的 Jute 最大缓冲区为 512 B，重启 ZK 和 Yarn 服务，此时 ZK 和 RM 服务均出现异常，ZK 异常信息表现为数据 java.io.IOException: Len error 614 客户端写入数据超过 512B 无法正常写入 ZK，RM 表现为 ”code:CONNECTIONLOSS“，无法连接到 ZK，两个 RM 均处于 Standy 状态，此时集群处于不可用状态。

leader ZK 异常信息：

2020-12-07 16:00:11,869 INFO org.apache.zookeeper.server.ZooKeeperServer: Client attempting to renew session 0x1763c3707800002 at /10.197.1.96:328922020-12-07 16:00:11,869 INFO org.apache.zookeeper.server.ZooKeeperServer: Established session 0x1763c3707800002 with negotiated timeout 40000 for client /10.197.1.96:328922020-12-07 16:00:11,870 WARN org.apache.zookeeper.server.NIOServerCnxn: Exception causing close of session 0x1763c3707800002 due to java.io.IOException: Len error 6142020-12-07 16:00:11,870 INFO org.apache.zookeeper.server.NIOServerCnxn: Closed socket connection for client /10.197.1.96:32892 which had sessionid 0x1763c37078000022020-12-07 16:00:12,216 INFO org.apache.zookeeper.server.NIOServerCnxnFactory: Accepted socket connection from /10.197.1.141:564922020-12-07 16:00:12,216 INFO org.apache.zookeeper.server.ZooKeeperServer: Client attempting to establish new session at /10.197.1.141:564922020-12-07 16:00:12,218 INFO org.apache.zookeeper.server.ZooKeeperServer: Established session 0x3763c3707830001 with negotiated timeout 40000 for client /10.197.1.141:564922020-12-07 16:00:12,219 WARN org.apache.zookeeper.server.NIOServerCnxn: Exception causing close of session 0x3763c3707830001 due to java.io.IOException: Len error 6142020-12-07 16:00:12,220 INFO org.apache.zookeeper.server.NIOServerCnxn: Closed socket connection for client /10.197.1.141:56492 which had sessionid 0x3763c37078300012020-12-07 16:00:14,275 INFO org.apache.zookeeper.server.NIOServerCnxnFactory: Accepted socket connection from /10.197.1.141:565102020-12-07 16:00:14,275 INFO org.apache.zookeeper.server.ZooKeeperServer: Client attempting to renew session 0x3763c3707830001 at /10.197.1.141:565102020-12-07 16:00:14,276 INFO org.apache.zookeeper.server.ZooKeeperServer: Established session 0x3763c3707830001 with negotiated timeout 40000 for client /10.197.1.141:565102020-12-07 16:00:14,276 WARN org.apache.zookeeper.server.NIOServerCnxn: Exception causing close of session 0x3763c3707830001 due to java.io.IOException: Len error 6142020-12-07 16:00:14,276 INFO org.apache.zookeeper.server.NIOServerCnxn: Closed socket connection for client /10.197.1.141:56510 which had sessionid 0x3763c37078300012020-12-07 16:00:16,000 INFO org.apache.zookeeper.server.ZooKeeperServer: Expiring session 0x1763c3707800000, timeout of 5000ms exceeded

复制代码

Yarn RM 日志：

2020-12-07 16:00:10,938 INFO org.apache.hadoop.ha.ActiveStandbyElector: Session connected.2020-12-07 16:00:10,938 INFO org.apache.hadoop.ha.ActiveStandbyElector: Ignore duplicate monitor lock-node request.2020-12-07 16:00:11,038 INFO org.apache.hadoop.ha.ActiveStandbyElector: Session disconnected. Entering neutral mode...2020-12-07 16:00:11,647 INFO org.apache.zookeeper.ClientCnxn: Opening socket connection to server slave-prd-10-197-1-236.v-bj-5.kwang.lan/10.197.1.236:2181. Will not attempt to authenticate using SASL (unknown error)2020-12-07 16:00:11,647 INFO org.apache.zookeeper.ClientCnxn: Socket connection established, initiating session, client: /10.197.1.141:56854, server: slave-prd-10-197-1-236.v-bj-5.kwang.lan/10.197.1.236:21812020-12-07 16:00:11,649 INFO org.apache.zookeeper.ClientCnxn: Session establishment complete on server slave-prd-10-197-1-236.v-bj-5.kwang.lan/10.197.1.236:2181, sessionid = 0x1763c3707800001, negotiated timeout = 400002020-12-07 16:00:11,649 INFO org.apache.hadoop.ha.ActiveStandbyElector: Session connected.2020-12-07 16:00:11,650 INFO org.apache.hadoop.ha.ActiveStandbyElector: Ignore duplicate monitor lock-node request.2020-12-07 16:00:11,650 INFO org.apache.zookeeper.ClientCnxn: Unable to read additional data from server sessionid 0x1763c3707800001, likely server has closed socket, closing socket connection and attempting reconnect2020-12-07 16:00:11,750 FATAL org.apache.hadoop.ha.ActiveStandbyElector: Received create error from Zookeeper. code:CONNECTIONLOSS for path /yarn-leader-election/yarnRM/ActiveStandbyElectorLock. Not retrying further znode create connection errors.2020-12-07 16:00:12,210 INFO org.apache.zookeeper.ZooKeeper: Session: 0x1763c3707800001 closed2020-12-07 16:00:12,212 WARN org.apache.hadoop.ha.ActiveStandbyElector: Ignoring stale result from old client with sessionId 0x1763c37078000012020-12-07 16:00:12,212 WARN org.apache.hadoop.ha.ActiveStandbyElector: Ignoring stale result from old client with sessionId 0x1763c37078000012020-12-07 16:00:12,212 INFO org.apache.zookeeper.ClientCnxn: EventThread shut down2020-12-07 16:00:12,213 ERROR org.apache.hadoop.yarn.server.resourcemanager.ResourceManager: Received RMFatalEvent of type EMBEDDED_ELECTOR_FAILED, caused by Received create error from Zookeeper. code:CONNECTIONLOSS for path /yarn-leader-election/yarnRM/ActiveStandbyElectorLock. Not retrying further znode create connection errors.2020-12-07 16:00:12,213 WARN org.apache.hadoop.yarn.server.resourcemanager.ResourceManager: Transitioning the resource manager to standby.2020-12-07 16:00:12,214 INFO org.apache.hadoop.yarn.server.resourcemanager.ResourceManager: Transitioning RM to Standby mode2020-12-07 16:00:12,214 INFO org.apache.hadoop.yarn.server.resourcemanager.ResourceManager: Already in standby state2020-12-07 16:00:12,214 INFO org.apache.hadoop.ha.ActiveStandbyElector: Yielding from electionÏ2020-12-07 16:00:12,214 INFO org.apache.zookeeper.ZooKeeper: Initiating client connection, connectString=slave-prd-10-197-1-236.v-bj-5.kwang.lan:2181,slave-prd-10-197-1-96.v-bj-5.kwang.lan:2181,slave-prd-10-197-1-141.v-bj-5.kwang.lan:2181 sessionTimeout=60000 watcher=org.apache.hadoop.ha.ActiveStandbyElector$WatcherWithClientRef@67b6359c2020-12-07 16:00:12,215 INFO org.apache.zookeeper.ClientCnxn: Opening socket connection to server slave-prd-10-197-1-141.v-bj-5.kwang.lan/10.197.1.141:2181. Will not attempt to authenticate using SASL (unknown error)2020-12-07 16:00:12,216 INFO org.apache.zookeeper.ClientCnxn: Socket connection established, initiating session, client: /10.197.1.141:56492, server: slave-prd-10-197-1-141.v-bj-5.kwang.lan/10.197.1.141:21812020-12-07 16:00:12,218 INFO org.apache.zookeeper.ClientCnxn: Session establishment complete on server slave-prd-10-197-1-141.v-bj-5.kwang.lan/10.197.1.141:2181, sessionid = 0x3763c3707830001, negotiated timeout = 400002020-12-07 16:00:12,219 INFO org.apache.hadoop.ha.ActiveStandbyElector: Session connected.2020-12-07 16:00:12,220 INFO org.apache.zookeeper.ClientCnxn: Unable to read additional data from server sessionid 0x3763c3707830001, likely server has closed socket, closing socket connection and attempting reconnect2020-12-07 16:00:12,320 INFO org.apache.hadoop.ha.ActiveStandbyElector: Session disconnected. Entering neutral mode...2020-12-07 16:00:12,320 WARN org.apache.hadoop.yarn.server.resourcemanager.EmbeddedElectorService: Lost contact with Zookeeper. Transitioning to standby in 60000 ms if connection is not reestablished.

复制代码

二、RM 与 ZNode 交互原理

2.1 RM 状态在 ZK 中的存储

不管 RM 是否启用了高可用，RM 作为 Yarn 的核心服务组件，不仅要与各个节点上的 ApplicationMaster 进行通信，还要与 NodeManager 进行心跳包的传输，自然在 RM 上会注册进来很多应用，每个应用由一个 ApplicationMaster 负责掌管整个应用周期，既然 RM 角色如此重要，就有必要保存一下 RM 的信息状态，以免 RM 进程异常退出后导致应用状态信息全部丢失，RM 重启无法重跑之前的任务。

既然应用状态信息要保存的目标易经明确了，那保存方式和保存的数据信息是什么呢。

在 Yarn 中 RM 应用状态信息保存的方式有四种：

MemoryRMStateStore——信息状态保存在内存中的实现类。
FileSystemRMStateStore——信息状态保存在 HDFS 文件系统中，这个是做了持久化的。
NullRMStateStore——什么都不做，就是不保存应用状态信息。
ZKRMStateStore——信息状态保存在 Zookeeper 中。

由于 Yarn 启用了 RM HA，以上四种方式只能支持 ZKRMStateStore。

那 RM 在 ZK 中到底是存储了哪些信息状态呢？如下所示，是 ZK 中存储 RM 信息状态的目录格式，可以看出，ZK 中主要存储 Application（作业的状态信息）和 SECRET_MANAGER（作业的 TOKEN 信息）等。

    ROOT_DIR_PATH      |--- VERSION_INFO      |--- EPOCH_NODE      |--- RM_ZK_FENCING_LOCK      |--- RM_APP_ROOT      |     |----- (#ApplicationId1)      |     |        |----- (#ApplicationAttemptIds)      |     |      |     |----- (#ApplicationId2)      |     |       |----- (#ApplicationAttemptIds)      |     ....      |      |--- RM_DT_SECRET_MANAGER_ROOT      |----- RM_DT_SEQUENTIAL_NUMBER_ZNODE_NAME      |----- RM_DELEGATION_TOKENS_ROOT_ZNODE_NAME      |       |----- Token_1      |       |----- Token_2      |       ....      |      |----- RM_DT_MASTER_KEYS_ROOT_ZNODE_NAME      |      |----- Key_1      |      |----- Key_2      ....      |--- AMRMTOKEN_SECRET_MANAGER_ROOT      |----- currentMasterKey      |----- nextMasterKey

复制代码

2.2 ZK 存储 &更新 RM 信息状态逻辑

作业提交到 Yarn 上的入口，都是通过 YarnClient 这个接口 api 提交的，具体提交方法为 submitApplication()。

//位置：org/apache/hadoop/yarn/client/api/YarnClient.java  public abstract ApplicationId submitApplication(      ApplicationSubmissionContext appContext) throws YarnException,      IOException;

复制代码

作业提交后，会经过一些列的事件转换，请求到不同的状态机进行处理，而保存作业的状态机 StoreAppTransition 会对 APP 的状态进行保存，将其元数据存储到 ZK 中。

//位置：org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStore.java  public void storeNewApplication(RMApp app) {    ApplicationSubmissionContext context = app                                            .getApplicationSubmissionContext();    assert context instanceof ApplicationSubmissionContextPBImpl;    ApplicationStateData appState =        ApplicationStateData.newInstance(            app.getSubmitTime(), app.getStartTime(), context, app.getUser());    // 向调度器发送 RMStateStoreEventType.STORE_APP 事件    dispatcher.getEventHandler().handle(new RMStateStoreAppEvent(appState));  }

复制代码

这里向调度器发送 RMStateStoreEventType.STORE_APP 事件，并注册了 StoreAppTransition 状态机。

//位置：org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStore.java    .addTransition(RMStateStoreState.ACTIVE,          EnumSet.of(RMStateStoreState.ACTIVE, RMStateStoreState.FENCED),          RMStateStoreEventType.STORE_APP, new StoreAppTransition())

复制代码

StoreAppTransition 状态机最终会调用 ZKRMStateStore#storeApplicationStateInternal() 方法，对 RM 的元数据在 ZK 中进行保存。

//位置：org/apache/hadoop/yarn/server/resourcemanager/recovery/ZKRMStateStore.java   @Override  public synchronized void storeApplicationStateInternal(ApplicationId appId,      ApplicationStateData appStateDataPB) throws Exception {    String nodeCreatePath = getNodePath(rmAppRoot, appId.toString());
    if (LOG.isDebugEnabled()) {      LOG.debug("Storing info for app: " + appId + " at: " + nodeCreatePath);    }    byte[] appStateData = appStateDataPB.getProto().toByteArray();	createWithRetries(nodeCreatePath, appStateData, zkAcl,              CreateMode.PERSISTENT);  }

复制代码

RM Application 的状态保存到 ZK 后，APP 状态最终会转化为 ACCETPED 状态，此时，会触发 StartAppAttemptTransition 状态机，对 AppAttemp 状态进行保存。

//位置：org/apache/hadoop/yarn/server/resourcemanager/recovery/ZKRMStateStore.java   @Override  public synchronized void storeApplicationAttemptStateInternal(      ApplicationAttemptId appAttemptId,      ApplicationAttemptStateData attemptStateDataPB)      throws Exception {    String appDirPath = getNodePath(rmAppRoot,        appAttemptId.getApplicationId().toString());    String nodeCreatePath = getNodePath(appDirPath, appAttemptId.toString());
    if (LOG.isDebugEnabled()) {      LOG.debug("Storing info for attempt: " + appAttemptId + " at: "          + nodeCreatePath);    }    byte[] attemptStateData = attemptStateDataPB.getProto().toByteArray();	createWithRetries(nodeCreatePath, attemptStateData, zkAcl,					CreateMode.PERSISTENT);  }

复制代码

而在任务运行结束时，会对 Application 和 AppAttemp 的状态进行更新。而更新操作也是容易出现异常的地方，这两段代码主要是执行更新或添加任务重试状态信息到 ZK 中的操作，Yarn 在调度任务的过程中，可能会对任务进行多次重试，主要受网络、硬件、资源等因素影响，如果任务重试信息保存在 ZK 失败，会调用 org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore.ZKAction.runWithRetries() 方法重试。

//位置：org/apache/hadoop/yarn/server/resourcemanager/recovery/ZKRMStateStore.java   // 对 Application 状态进行更新  @Override  public synchronized void updateApplicationStateInternal(ApplicationId appId,      ApplicationStateData appStateDataPB) throws Exception {    String nodeUpdatePath = getNodePath(rmAppRoot, appId.toString());
    if (LOG.isDebugEnabled()) {      LOG.debug("Storing final state info for app: " + appId + " at: "          + nodeUpdatePath);    }    byte[] appStateData = appStateDataPB.getProto().toByteArray();
    if (existsWithRetries(nodeUpdatePath, false) != null) {      setDataWithRetries(nodeUpdatePath, appStateData, -1);    } else {      createWithRetries(nodeUpdatePath, appStateData, zkAcl,              CreateMode.PERSISTENT);      LOG.debug(appId + " znode didn't exist. Created a new znode to"              + " update the application state.");    }  }
  // 对 AppAttemp 状态进行更新  @Override  public synchronized void updateApplicationAttemptStateInternal(      ApplicationAttemptId appAttemptId,      ApplicationAttemptStateData attemptStateDataPB)      throws Exception {    String appIdStr = appAttemptId.getApplicationId().toString();    String appAttemptIdStr = appAttemptId.toString();    String appDirPath = getNodePath(rmAppRoot, appIdStr);    String nodeUpdatePath = getNodePath(appDirPath, appAttemptIdStr);    if (LOG.isDebugEnabled()) {      LOG.debug("Storing final state info for attempt: " + appAttemptIdStr          + " at: " + nodeUpdatePath);    }    byte[] attemptStateData = attemptStateDataPB.getProto().toByteArray();
    if (existsWithRetries(nodeUpdatePath, false) != null) {      setDataWithRetries(nodeUpdatePath, attemptStateData, -1);    } else {      createWithRetries(nodeUpdatePath, attemptStateData, zkAcl,              CreateMode.PERSISTENT);      LOG.debug(appAttemptId + " znode didn't exist. Created a new znode to"              + " update the application attempt state.");    }  }

复制代码

在启用 Yarn 高可用情况下，

重试间隔机制如下：受 yarn.resourcemanager.zk-timeout-ms（ZK 会话超时时间，线上 1 分钟，即 60000ms）和 yarn.resourcemanager.zk-num-retries（操作失败后重试次数，线上环境 1000 次）参数控制，计算公式为：

重试时间间隔（yarn.resourcemanager.zk-retry-interval-ms ）=yarn.resourcemanager.zk-timeout-ms(ZK session超时时间)/yarn.resourcemanager.zk-num-retries（重试次数）

复制代码

即在生产环境中，重试时间间隔 = 600000ms /1000 次 = 60 ms/次，即线上环境在任务不成功的条件下，会重试 1000 次，每次 60 ms，这里也可能会导致 RM 堆内存溢出。参考资料：https://my.oschina.net/dabird/blog/3089265。

重试间隔确定代码如下：

//位置：src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/ZKRMStateStore.java    @Override  public synchronized void initInternal(Configuration conf) throws Exception {    zkHostPort = conf.get(YarnConfiguration.RM_ZK_ADDRESS);    if (zkHostPort == null) {      throw new YarnRuntimeException("No server address specified for " +          "zookeeper state store for Resource Manager recovery. " +          YarnConfiguration.RM_ZK_ADDRESS + " is not configured.");    }    // ZK 连接重试次数    numRetries =        conf.getInt(YarnConfiguration.RM_ZK_NUM_RETRIES,            YarnConfiguration.DEFAULT_ZK_RM_NUM_RETRIES);    znodeWorkingPath =        conf.get(YarnConfiguration.ZK_RM_STATE_STORE_PARENT_PATH,            YarnConfiguration.DEFAULT_ZK_RM_STATE_STORE_PARENT_PATH);
    // ZK session 超时时间    zkSessionTimeout =        conf.getInt(YarnConfiguration.RM_ZK_TIMEOUT_MS,            YarnConfiguration.DEFAULT_RM_ZK_TIMEOUT_MS);    zknodeLimit =        conf.getInt(YarnConfiguration.RM_ZK_ZNODE_SIZE_LIMIT_BYTES,            YarnConfiguration.DEFAULT_RM_ZK_ZNODE_SIZE_LIMIT_BYTES);
    if (HAUtil.isHAEnabled(conf)) {      zkRetryInterval = zkSessionTimeout / numRetries;    } else {      zkRetryInterval =          conf.getLong(YarnConfiguration.RM_ZK_RETRY_INTERVAL_MS,              YarnConfiguration.DEFAULT_RM_ZK_RETRY_INTERVAL_MS);    } }

复制代码

至此，我们已经清楚了 RM 中作业的信息状态是如何保存在 ZK 中并如何进行更新的。

2.3 ZK 删除 RM 信息状态逻辑

在了解了 RM 作业信息状态保存在 ZK 的逻辑后，我们便会产生一个疑问，那 RM 状态保存在 ZK 中后，是否会一直驻留在 ZK 中呢？答案是否定的，ZK 也会对作业的状态进行删除，那删除逻辑是这样的呢？

删除的核心逻辑位于 RMAppManager#checkAppNumCompletedLimit() 方法中调用的 removeApplication() 方法，其逻辑就是判断保存在 ZK StateStore 中或已完成的作业数量超过对应限制，则对 App 状态信息进行删除。

//位置：org/apache/hadoop/yarn/server/resourcemanager/RMAppManager.java  /*   * check to see if hit the limit for max # completed apps kept   */  protected synchronized void checkAppNumCompletedLimit() {    // check apps kept in state store.    while (completedAppsInStateStore > this.maxCompletedAppsInStateStore) {      ApplicationId removeId =          completedApps.get(completedApps.size() - completedAppsInStateStore);      RMApp removeApp = rmContext.getRMApps().get(removeId);      LOG.info("Max number of completed apps kept in state store met:"          + " maxCompletedAppsInStateStore = " + maxCompletedAppsInStateStore          + ", removing app " + removeApp.getApplicationId()          + " from state store.");      rmContext.getStateStore().removeApplication(removeApp);      completedAppsInStateStore--;    }
    // check apps kept in memorty.    while (completedApps.size() > this.maxCompletedAppsInMemory) {      ApplicationId removeId = completedApps.remove();      LOG.info("Application should be expired, max number of completed apps"          + " kept in memory met: maxCompletedAppsInMemory = "          + this.maxCompletedAppsInMemory + ", removing app " + removeId          + " from memory: ");      rmContext.getRMApps().remove(removeId);      this.applicationACLsManager.removeApplication(removeId);    }  }

复制代码

可以看看相关参数是如何设置的，其中保存在 ZK StateStore 中和保存在 Memory 的 App 最大数量是一致的，默认是 10000（线上环境默认也是 10000），且保存在 ZK StateSotre 中的作业数量不能超过保存在 Memory 中的作业数量。

//位置：org/apache/hadoop/yarn/server/resourcemanager/RMAppManager.java  public RMAppManager(RMContext context,      YarnScheduler scheduler, ApplicationMasterService masterService,      ApplicationACLsManager applicationACLsManager, Configuration conf) {    ...    // 保存在 Memory 中的 App 最大数量    this.maxCompletedAppsInMemory = conf.getInt(        YarnConfiguration.RM_MAX_COMPLETED_APPLICATIONS,        YarnConfiguration.DEFAULT_RM_MAX_COMPLETED_APPLICATIONS);    // 保存在 ZK StateStore 中的 App 最大数量，默认和 Memory 中的最大值保存一致    this.maxCompletedAppsInStateStore =        conf.getInt(          YarnConfiguration.RM_STATE_STORE_MAX_COMPLETED_APPLICATIONS,          YarnConfiguration.DEFAULT_RM_STATE_STORE_MAX_COMPLETED_APPLICATIONS);
    // 保存在 ZK StateStore 中的 App 数量不能超过保存在 Memory 中的 App 数量    if (this.maxCompletedAppsInStateStore > this.maxCompletedAppsInMemory) {      this.maxCompletedAppsInStateStore = this.maxCompletedAppsInMemory;    }  }
//位置：org/apache/hadoop/yarn/conf/YarnConfiguration.java  // maxCompletedAppsInMemory 参数定义  /** The maximum number of completed applications RM keeps. */   public static final String RM_MAX_COMPLETED_APPLICATIONS =    RM_PREFIX + "max-completed-applications";  public static final int DEFAULT_RM_MAX_COMPLETED_APPLICATIONS = 10000;
  // maxCompletedAppsInStateStore 参数定义，默认和 maxCompletedAppsInMemory 保持一致  /**   * The maximum number of completed applications RM state store keeps, by   * default equals to DEFAULT_RM_MAX_COMPLETED_APPLICATIONS   */  public static final String RM_STATE_STORE_MAX_COMPLETED_APPLICATIONS =      RM_PREFIX + "state-store.max-completed-applications";  public static final int DEFAULT_RM_STATE_STORE_MAX_COMPLETED_APPLICATIONS =      DEFAULT_RM_MAX_COMPLETED_APPLICATIONS;

复制代码

执行真正的删除操作，删除在 ZK 中保存的超出限制的 App 状态信息。

//位置：org/apache/hadoop/yarn/server/resourcemanager/recovery/ZKRMStateStore.java   @Override  public synchronized void removeApplicationStateInternal(      ApplicationStateData  appState)      throws Exception {    String appId = appState.getApplicationSubmissionContext().getApplicationId()        .toString();    String appIdRemovePath = getNodePath(rmAppRoot, appId);    ArrayList<Op> opList = new ArrayList<Op>();
    // 删除在 ZK 中保存的 AppAttempt 信息    for (ApplicationAttemptId attemptId : appState.attempts.keySet()) {      String attemptRemovePath = getNodePath(appIdRemovePath, attemptId.toString());      opList.add(Op.delete(attemptRemovePath, -1));    }    opList.add(Op.delete(appIdRemovePath, -1));
    if (LOG.isDebugEnabled()) {      LOG.debug("Removing info for app: " + appId + " at: " + appIdRemovePath          + " and its attempts.");    }    // 删除在 ZK 中保存的 Applicaton 信息    doDeleteMultiWithRetries(opList);  }

复制代码

三、解决方案

3.1 Hadoop 2.9.0 之前修复方法

RM 状态在 ZK 存储的过程中，RM 作为客户端，ZK 作为服务端，在 Hadoop 2.9.0 版本之前，出现这种异常的处理方式为修改 ZK 端 jute.maxbuffer 参数的值，以增加 RM 作业允许写 ZK 的最大值。但这种处理方式有三种不足：

ZK 服务端允许写入的 ZNode 数据量太大，会影响 ZK 服务的读写性能和 ZK 内存紧张；
需要重启 ZK 服务端和客户端 RM 服务，运维成本较高。（如果有其他服务依赖此 ZK 则成本更高，可能还需要重启其他服务）
异常任务写 ZNode 数据量不可控，某些情况下还是会发生写入 ZNode 大小超过限制。

Q：为什么要限制 ZK 中 ZNode 大小？

A：ZK 是一套高吞吐量的系统，为了提高系统的读取速度，ZK 不允许从文件中读取需要的数据，而是直接从内存中查找。换句话说，ZK 集群中每一台服务器都包含全量的数据，并且这些数据都会加载到内存中，同时 ZNode 的数据不支持 Append 操作，全部都是 Replace 操作。如果 ZNode 数据量过大，那么读写 ZNode 将造成不确定的延时（比如服务端同步数据慢），同时 ZNode 太大会消耗 ZK 服务器的内存，这也是为什么 ZK 不适合存储大量数据的原因。

3.2 Hadoop 2.9.0 及后续版本修复方法

在 Hadoop 2.9.0 及后续版本中，yarn-site.xml 中增加了 yarn.resourcemanager.zk-max-znode-size.bytes 参数，该参数定义了 ZK 的 ZNode 节点所能存储的最大数据量，以字节为单位，默认是 1024*1024 字节，也就是 1MB。使用这种方式，我们就不需要修改 ZK 的服务端的配置，而只需修改 Yarn 服务端的配置并重启 RM 服务，就能限制 RM 往 ZK 中写入的数据量，而且也提高了 ZK 服务的可用性。

修复的核心主要是在 ZKRMStateStore 类中的 storeApplicationStateInternal()、updateApplicationStateInternal()、storeApplicationAttemptStateInternal()、updateApplicationAttemptStateInternal() 方法逻辑中增加了是否超过写 ZNode 大小限制的判断，避免单个作业写 ZNode 数据量过大导致 RM 和 ZK 服务的不可用。部分代码如下：

//位置：org/apache/hadoop/yarn/server/resourcemanager/recovery/ZKRMStateStore.java   // Application 写 ZNode 时判断大小限制    @Override  public synchronized void storeApplicationStateInternal(ApplicationId appId,      ApplicationStateData appStateDataPB) throws Exception {    String nodeCreatePath = getNodePath(rmAppRoot, appId.toString());
    if (LOG.isDebugEnabled()) {      LOG.debug("Storing info for app: " + appId + " at: " + nodeCreatePath);    }    byte[] appStateData = appStateDataPB.getProto().toByteArray();    if (appStateData.length <= zknodeLimit) {      createWithRetries(nodeCreatePath, appStateData, zkAcl,              CreateMode.PERSISTENT);      LOG.debug("Store application state data size for " + appId + " is " + appStateData.length);    } else {      LOG.info("Store application state data size for " + appId + " is " + appStateData.length +        ". exceeds the maximum allowed size " + zknodeLimit + " for application data.");    }  }
  // Application 状态更新时判断写 ZNode 大小  @Override  public synchronized void updateApplicationStateInternal(ApplicationId appId,      ApplicationStateData appStateDataPB) throws Exception {    String nodeUpdatePath = getNodePath(rmAppRoot, appId.toString());
    if (LOG.isDebugEnabled()) {      LOG.debug("Storing final state info for app: " + appId + " at: "          + nodeUpdatePath);    }    byte[] appStateData = appStateDataPB.getProto().toByteArray();
    if (appStateData.length <= zknodeLimit) {      if (existsWithRetries(nodeUpdatePath, false) != null) {        setDataWithRetries(nodeUpdatePath, appStateData, -1);      } else {        createWithRetries(nodeUpdatePath, appStateData, zkAcl,                CreateMode.PERSISTENT);        LOG.debug(appId + " znode didn't exist. Created a new znode to"                + " update the application state.");      }      LOG.debug("Update application state data size for " + appId + " is " + appStateData.length);    } else {      LOG.info("Update application state data size for " + appId + " is " + appStateData.length +              ". exceeds the maximum allowed size " + zknodeLimit + " for application data.");    }  }

复制代码

3.3 任务测试

设置 Yarn app 允许写 ZNode 的最大值，重启 active RM

参数：yarn-site.xml 的 ResourceManager 高级配置代码段（安全阀）值：<property>    <name>yarn.resourcemanager.zk-max-znode-size.bytes</name>    <value>512</value></property>

复制代码

测试任务：

hadoop jar /opt/cloudera/parcels/CDH-5.14.4-1.cdh5.14.4.p0.3/jars/hadoop-mapreduce-examples-2.6.0-cdh5.14.4.jar  pi -Dmapred.job.queue.name=root.exquery 20 10

复制代码

任务失败时 RM 任务日志如下，可以看出作业状态信息保存在 ZK 的数据超过了 ZNode 限制，此时 ZK 不会保存该作业的状态信息，而 ZK 服务和 RM 服务均是正常对外提供服务的，不影响集群的正常使用。

# tailf hadoop-cmf-yarn-RESOURCEMANAGER-slave-prd-10-197-1-141.v-bj-5.kwang.lan.log.out  |grep "the maximum allowed size"2020-12-10 16:53:37,544 INFO org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore: Application state data size for application_1607589684539_0001 is 1515. exceeds the maximum allowed size 512 for application data.2020-12-10 16:53:48,086 INFO org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore: Application state data size for application_1607590418121_0001 is 1515. exceeds the maximum allowed size 512 for application data.
# RM 具体 Warn 信息：2020-12-10 16:53:49,377 WARN org.apache.hadoop.security.UserGroupInformation: PriviledgedActionException as:kwang (auth:SIMPLE) cause:org.apache.hadoop.yarn.exceptions.ApplicationNotFoundException: Application with id 'application_1607590418121_0001' doesn't exist in RM.2020-12-10 16:53:49,377 INFO org.apache.hadoop.ipc.Server: IPC Server handler 0 on 8032, call org.apache.hadoop.yarn.api.ApplicationClientProtocolPB.getApplicationReport from 10.197.1.141:56026 Call#63 Retry#0org.apache.hadoop.yarn.exceptions.ApplicationNotFoundException: Application with id 'application_1607590418121_0001' doesn't exist in RM.        at org.apache.hadoop.yarn.server.resourcemanager.ClientRMService.getApplicationReport(ClientRMService.java:324)        at org.apache.hadoop.yarn.api.impl.pb.service.ApplicationClientProtocolPBServiceImpl.getApplicationReport(ApplicationClientProtocolPBServiceImpl.java:170)        at org.apache.hadoop.yarn.proto.ApplicationClientProtocol$ApplicationClientProtocolService$2.callBlockingMethod(ApplicationClientProtocol.java:401)        at org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:617)        at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:1073)        at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2281)        at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2277)        at java.security.AccessController.doPrivileged(Native Method)        at javax.security.auth.Subject.doAs(Subject.java:422)        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1924)        at org.apache.hadoop.ipc.Server$Handler.run(Server.java:2275)

复制代码

四、修复方法

4.1 修复 patch 并变更参数

修复方式参考参考资料1，并设置 Yarn app 允许写 ZNode 的最大值（4*1024*1024 B，即 4M），重启 RM。

参数：yarn-site.xml 的 ResourceManager 高级配置代码段（安全阀）值：<property>    <name>yarn.resourcemanager.zk-max-znode-size.bytes</name>    <value>4194304</value></property>

复制代码

4.2 建议参数变更

前面在 2.2 小节中分析了作业在更新 Application 或 AppAttemp 状态时，会通过重试的方式向 ZK 的 ZNode 中写入数据，线上环境默认的重试次数为 1000 次，重试间隔为 60ms，而一旦任务出现异常时，这种高频次的写入会对 ZK 或 RM 服务造成一定的压力，因此可以调小作业的重试次数，减少重试时对服务的压力。

参数：yarn-site.xml 的 ResourceManager 高级配置代码段（安全阀）值：<property>    <name>yarn.resourcemanager.zk-num-retries</name>    <value>100</value></property>

复制代码

【参考资料】

发布于: 2021 年 01 月 08 日阅读数: 42

原文链接:【http://xie.infoq.cn/article/ec724273849384e2fb2dd3bf4】。文章转载请联系作者。

笨小康

关注

万物之中，希望至美！ 2018.09.17 加入

还未添加个人简介

发布

暂无评论

创作场景

Yarn RM 写 ZNode 超数据量限制 bug 修复

问题背景

一、问题复现

二、RM 与 ZNode 交互原理

2.1 RM 状态在 ZK 中的存储

2.2 ZK 存储 &更新 RM 信息状态逻辑

2.3 ZK 删除 RM 信息状态逻辑

三、解决方案

3.1 Hadoop 2.9.0 之前修复方法

3.2 Hadoop 2.9.0 及后续版本修复方法

3.3 任务测试

四、修复方法

4.1 修复 patch 并变更参数

4.2 建议参数变更

【参考资料】

笨小康

评论