hadoop Yarn的 RMStateStore对比分析

hdfs://namenode:8020/rmstate/FSRMStateRoot/
├── RMVersionNode                    # 版本信息
├── EpochNode                       # Epoch计数器
├── RMDTSecretManagerRoot/         # 委托令牌
│   ├── DelegationKey_<id>
│   ├── RMDelegationToken_<seq>
│   └── RMDTSequenceNumber_<seq>
├── AMRMTokenSecretManagerRoot/    # AMRM令牌
│   └── AMRMTokenSecretManagerNode
├── RMAppRoot/                     # 应用状态
│   └── <ApplicationId>/
│       ├── <ApplicationId>
│       └── attempt_<id>
├── ReservationSystemRoot/         # 资源预留
│   └── <PlanName>/
└── ProxyCARoot/                   # CA证书
    ├── caCert
    └── caPrivateKey

数据库: yarn-rm-state (LevelDB)

Key                                    | Value
--------------------------------------|------------------------------
RMVersionNode                          | VersionProto
EpochNode                              | EpochProto
RMDTSecretManagerRoot/DelegationKey_<id> | DelegationKey
RMDTSecretManagerRoot/RMDelegationToken_<seq> | TokenData
RMDTSecretManagerRoot/RMDTSequentialNumber | Int
RMAppRoot/application_<id>           | ApplicationStateDataProto
AMRMTokenSecretManagerRoot            | AMRMTokenStateProto
ProxyCARoot/caCert                    | X509Certificate (DER)
ProxyCARoot/caPrivateKey              | PrivateKey (DER)
ReservationSystemRoot/<plan>/<res_id> | ReservationProto

/yarn rm/ZKRMStateRoot/
├── VERSION_INFO                      # 版本
├── EPOCH_NODE                       # Epoch
├── RM_ZK_FENCING_LOCK              # 锁节点 (HA)
├── RMAppRoot/
│   ├── HIERARCHIES/                # 分层索引 (1-4)
│   │   ├── 1/application_1
│   │   │   └── application_1234_0001_01  # 尝试状态
│   │   ├── 2/application_12
│   │   │   └── application_34_0001_01
│   │   └── ...
│   └── application_<id>
│       └── attempt_<id>
├── RMDTSecretManagerRoot/
│   ├── RMDTSequentialNumber        # 序列号
│   ├── RMDTMasterKeysRoot/
│   │   └── Key_<id>
│   └── RMDelegationTokensRoot/
│       ├── 1/Token_1
│       │   └── Token_1234
│       └── ...
├── AMRMTokenSecretManagerRoot/
│   ├── currentMasterKey
│   └── nextMasterKey
├── ReservationSystemRoot/
│   └── <PlanName>/
│       └── <ReservationId>
└── ProxyCARoot/
    ├── caCert
    └── caPrivateKey

// FileSystemRMStateStore - 临时文件策略
protected void writeFile(Path outputPath, byte[] data, boolean makeUnreadableByAdmin) {
    Path tempPath = new Path(outputPath.getParent(), outputPath.getName() + ".tmp");
    fsOut = fs.create(tempPath, true);
    fsOut.write(data);
    fsOut.close();
    fs.rename(tempPath, outputPath);  // 原子重命名
}

// LeveldbRMStateStore - 直接写入
protected void storeApplicationStateInternal(ApplicationId appId, ApplicationStateData appStateData) {
    db.put(bytes(key), appStateData.getProto().toByteArray());
}

// ZKRMStateStore - 安全创建
public synchronized void storeApplicationStateInternal(ApplicationId appId, 
    ApplicationStateData appStateDataPB) throws Exception {
    String nodeCreatePath = getLeafAppIdNodePath(appId.toString(), true);
    zkManager.safeCreate(nodeCreatePath, appStateData, zkAcl,
        CreateMode.PERSISTENT, zkAcl, fencingNodePath);
}

// FileSystemRMStateStore - 分开存储，存在不一致风险
private void storeOrUpdateRMDelegationTokenState(...) {
    writeFileWithRetries(nodeCreatePath, identifierData.toByteArray(), true);
    // 序列号单独存储...
    if (dtSequenceNumberPath == null) {
        createFileWithRetries(latestSequenceNumberPath);
    } else {
        renameFileWithRetries(dtSequenceNumberPath, latestSequenceNumberPath);
    }
}

// LeveldbRMStateStore - 原子批量操作
private void storeOrUpdateRMDT(...) {
    try (WriteBatch batch = db.createWriteBatch()) {
        batch.put(bytes(tokenKey), tokenData.toByteArray());
        if (!isUpdate) {
            batch.put(bytes(RM_DT_SEQUENCE_NUMBER_KEY), bs.toByteArray());
        }
        db.write(batch);  // 原子提交
    }
}

// ZKRMStateStore - ZK 事务
protected synchronized void storeRMDelegationTokenState(...) {
    SafeTransaction trx = zkManager.createTransaction(zkAcl, fencingNodePath);
    trx.create(nodeCreatePath, identifierData.toByteArray(), zkAcl, CreateMode.PERSISTENT);
    trx.setData(dtSequenceNumberPath, seqOs.toByteArray(), -1);
    trx.commit();  // 原子提交
}

// Fencing 锁节点
private static final String FENCING_LOCK = "RM_ZK_FENCING_LOCK";
private String fencingNodePath;

// 验证活跃状态线程
private class VerifyActiveStatusThread extends Thread {
    @Override
    public void work() {
        while (!isFencedState()) {
            // 周期性创建/删除 fence 节点
            zkManager.createTransaction(zkAcl, fencingNodePath).commit();
            Thread.sleep(zkSessionTimeout);
        }
    }
}

// 安全事务 - 带 fencing
public void safeCreate(String path, ...) {
    SafeTransaction trx = zkManager.createTransaction(zkAcl, fencingNodePath);
    trx.create(path, data, acl, mode);
    trx.commit();
}

// 应用ID分层存储 - 避免单节点children过多
// 配置: yarn.resourcemanager.zk-appid-node-split-index
// splitIndex=2 时: application_12 / application_34_0001

private Map<Integer, String> rmAppRootHierarchies;
for (int splitIndex = 1; splitIndex <= 4; splitIndex++) {
    rmAppRootHierarchies.put(splitIndex,
        getNodePath(hierarchiesPath, Integer.toString(splitIndex)));
}

<property>
    <name>yarn.resourcemanager.store.class</name>
    <value>org.apache.hadoop.yarn.server.resourcemanager.recovery.FileSystemRMStateStore</value>
</property>
<property>
    <name>yarn.resourcemanager.fs.state-store.uri</name>
    <value>hdfs://namenode:8020/rmstate</value>
</property>
<property>
    <name>yarn.resourcemanager.fs.state-store.num-retries</name>
    <value>15</value>
</property>
<property>
    <name>yarn.resourcemanager.fs.state-store.retry-interval-ms</name>
    <value>1000</value>
</property>

<property>
    <name>yarn.resourcemanager.store.class</name>
    <value>org.apache.hadoop.yarn.server.resourcemanager.recovery.LeveldbRMStateStore</value>
</property>
<property>
    <name>yarn.resourcemanager.leveldb.state-store.path</name>
    <value>/var/log/yarn/rmstate</value>
</property>
<property>
    <name>yarn.resourcemanager.leveldb.compaction-interval-secs</name>
    <value>3600</value>
</property>

<property>
    <name>yarn.resourcemanager.store.class</name>
    <value>org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore</value>
</property>
<property>
    <name>yarn.resourcemanager.zk.state-store.parent-path</name>
    <value>/yarn rm</value>
</property>
<property>
    <name>yarn.resourcemanager.zk.timeout-ms</name>
    <value>10000</value>
</property>
<property>
    <name>yarn.resourcemanager.zk.num-retries</name>
    <value>3</value>
</property>
<property>
    <name>yarn.resourcemanager.zk.znode-size-limit-bytes</name>
    <value>10485760</value>
</property>

                        场景选择决策树
                                    │
                                    ▼
                    ┌───────────────────────────┐
                    │     是否有 HA 需求？       │
                    └───────────────────────────┘
                         /               \
                       是                 否
                       /                   \
            ┌─────────────────┐    ┌─────────────────┐
            │  是否有共享存储？ │    │  性能要求如何？  │
            └─────────────────┘    └─────────────────┘
              /           \           /           \
            是           否         高             低
            /             \         /              \
┌──────────────────┐  ┌──────────┐┌──────────┐ ┌──────────────┐
│ FileSystemRMState│  │ ZKRM    ││ Leveldb  │ │ MemoryRM     │
│ Store (HDFS)     │  │StateStore││StateStore│ │ StateStore   │
└──────────────────┘  └──────────┘└──────────┘ └──────────────┘