本文基于 Hadoop3.1.2版本講解
HDFS 檔案洗掉程序
下面是hdfs洗掉路徑的方法，原始碼路徑org.apache.hadoop.hdfs.server.namenode.FSNamesystem.delete()
/**
* Remove the indicated file from namespace.
*
* @see ClientProtocol#delete(String, boolean) for detailed description and
* description of exceptions
*/
boolean delete(String src, boolean recursive, boolean logRetryCache)
throws IOException {
waitForLoadingFSImage();
BlocksMapUpdateInfo toRemovedBlocks = null;
writeLock();
boolean ret = false;
try {
checkOperation(OperationCategory.WRITE);
checkNameNodeSafeMode("Cannot delete " + src);
toRemovedBlocks = FSDirDeleteOp.delete( // @1
this, src, recursive, logRetryCache);
ret = toRemovedBlocks != null;
} catch (AccessControlException e) {
logAuditEvent(false, "delete", src);
throw e;
} finally {
writeUnlock();
}
getEditLog().logSync();
if (toRemovedBlocks != null) {
removeBlocks(toRemovedBlocks); // @2
}
logAuditEvent(true, "delete", src);
return ret;
}
代碼@1里面做了三件比較重要的事
①從 NameNode 維護的的目錄樹里面洗掉路徑，這也是為什么執行洗掉操作之后就無法再通過hdfs dfs -ls xxx 或其它 api 方式再查看到路徑的根本原因（拒絕需要被洗掉的檔案的外部訪問）
②找出被刪路徑關聯的 block 資訊，每個檔案包含多個 block 塊，分布在各個 DataNode,此時并未真正物理洗掉 DataNode 上物理磁盤上的block塊
③記錄洗掉日志到editlog（這一步也很重要，甚至是后面恢復資料的關鍵）
代碼@2把將要洗掉的block資訊添加到org.apache.hadoop.hdfs.server.blockmanagement.BlockManager 里面維護的 InvalidateBlocks 物件中，InvalidateBlocks 專門用于保存等待洗掉的資料塊副本
以上步驟并未涉及真正的物理洗掉的操作
BlockManager
BlockManager 管理了hdfs block 的生命周期并且維護在 Hadoop 集群中的塊相關的資訊，包括快的上報、復制、洗掉、監控、標記等等一系列功能，
BlockManager 中有個方法 invalidateWorkForOneNode() 專門用于定時洗掉 InvalidateBlocks 中存盤的待洗掉的快，此方法會在 NameNode 啟動時在 BlockManager 的內部執行緒類ReplicationMonitor 定時輪循把要洗掉的塊放入 DatanodeDescriptor 中的邏輯，方法的呼叫路徑如下：
org.apache.hadoop.hdfs.server.namenode.NameNode#initialize(Configuration conf)
org.apache.hadoop.hdfs.server.namenode.NameNode#startCommonServices(Configuration conf)
org.apache.hadoop.hdfs.server.namenode.FSNamesystem#startCommonServices(Configuration conf, HAContext haContext)
org.apache.hadoop.hdfs.server.blockmanagement.BlockManager#activate(Configuration conf)
org.apache.hadoop.hdfs.server.blockmanagement.BlockManager.ReplicationMonitor#run()
org.apache.hadoop.hdfs.server.blockmanagement.BlockManager#computeDatanodeWork()
org.apache.hadoop.hdfs.server.blockmanagement.BlockManager#computeDatanodeWorkcomputeInvalidateWork(int nodesToProcess)
org.apache.hadoop.hdfs.server.blockmanagement.BlockManager#invalidateWorkForOneNode(DatanodeInfo dn)
org.apache.hadoop.hdfs.server.blockmanagement.InvalidateBlocks#invalidateWork(final DatanodeDescriptor dn)
BlockManager 維護了 InvalidateBlocks，存放了待洗掉的 block，BlockManager 在 NameNode 啟動時會單獨啟動一個執行緒，定時把要洗掉的塊資訊放入 InvalidateBlocks 中，每次會從InvalidateBlocks 佇列中為每個 DataNode 取出 blockInvalidateLimit（由配置項dfs.block.invalidate.limit，默認1000）個塊邏輯在 BlockManager.computeInvalidateWork() 方法里會把要洗掉的塊資訊放入 DatanodeDescriptor 中的 invalidateBlocks 陣列，DatanodeManager 再通過 DataNode 與NameNode 心跳時，構建洗掉塊的指令集，NameNode 再把指令下發給DataNode，心跳由 DatanodeProtocol 呼叫，方法的呼叫路徑如下：
org.apache.hadoop.hdfs.server.protocol.DatanodeProtocol#sendHeartbeat()
org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer#sendHeartbeat()
org.apache.hadoop.hdfs.server.namenode.FSNamesystem#handleHeartbeat()
org.apache.hadoop.hdfs.server.blockmanagement.DatanodeManager#handleHeartbeat()
DatanodeManager.handleHeartbeat() 中構建洗掉的指令給 DataNode，待 NameNode 發送的代碼如下：
/** Handle heartbeat from datanodes. */
public DatanodeCommand[] handleHeartbeat(DatanodeRegistration nodeReg,
StorageReport[] reports, final String blockPoolId,
long cacheCapacity, long cacheUsed, int xceiverCount,
int maxTransfers, int failedVolumes,
VolumeFailureSummary volumeFailureSummary,
@Nonnull SlowPeerReports slowPeers,
@Nonnull SlowDiskReports slowDisks) throws IOException {
final DatanodeDescriptor nodeinfo;
try {
nodeinfo = getDatanode(nodeReg);
} catch (UnregisteredNodeException e) {
return new DatanodeCommand[]{RegisterCommand.REGISTER};
}
// Check if this datanode should actually be shutdown instead.
if (nodeinfo != null && nodeinfo.isDisallowed()) {
setDatanodeDead(nodeinfo);
throw new DisallowedDatanodeException(nodeinfo);
}
if (nodeinfo == null || !nodeinfo.isRegistered()) {
return new DatanodeCommand[]{RegisterCommand.REGISTER};
}
heartbeatManager.updateHeartbeat(nodeinfo, reports, cacheCapacity,
cacheUsed, xceiverCount, failedVolumes, volumeFailureSummary);
// If we are in safemode, do not send back any recovery / replication
// requests. Don't even drain the existing queue of work.
if (namesystem.isInSafeMode()) {
return new DatanodeCommand[0];
}
// block recovery command
final BlockRecoveryCommand brCommand = getBlockRecoveryCommand(blockPoolId,
nodeinfo);
if (brCommand != null) {
return new DatanodeCommand[]{brCommand};
}
final List<DatanodeCommand> cmds = new ArrayList<>();
// Allocate _approximately_ maxTransfers pending tasks to DataNode.
// NN chooses pending tasks based on the ratio between the lengths of
// replication and erasure-coded block queues.
int totalReplicateBlocks = nodeinfo.getNumberOfReplicateBlocks();
int totalECBlocks = nodeinfo.getNumberOfBlocksToBeErasureCoded();
int totalBlocks = totalReplicateBlocks + totalECBlocks;
if (totalBlocks > 0) {
int numReplicationTasks = (int) Math.ceil(
(double) (totalReplicateBlocks * maxTransfers) / totalBlocks);
int numECTasks = (int) Math.ceil(
(double) (totalECBlocks * maxTransfers) / totalBlocks);
if (LOG.isDebugEnabled()) {
LOG.debug("Pending replication tasks: " + numReplicationTasks
+ " erasure-coded tasks: " + numECTasks);
}
// check pending replication tasks
List<BlockTargetPair> pendingList = nodeinfo.getReplicationCommand(
numReplicationTasks);
if (pendingList != null && !pendingList.isEmpty()) {
cmds.add(new BlockCommand(DatanodeProtocol.DNA_TRANSFER, blockPoolId,
pendingList));
}
// check pending erasure coding tasks
List<BlockECReconstructionInfo> pendingECList = nodeinfo
.getErasureCodeCommand(numECTasks);
if (pendingECList != null && !pendingECList.isEmpty()) {
cmds.add(new BlockECReconstructionCommand(
DNA_ERASURE_CODING_RECONSTRUCTION, pendingECList));
}
}
// check block invalidation
Block[] blks = nodeinfo.getInvalidateBlocks(blockInvalidateLimit);
if (blks != null) {
cmds.add(new BlockCommand(DatanodeProtocol.DNA_INVALIDATE, blockPoolId,
blks));
}
// cache commands
addCacheCommands(blockPoolId, nodeinfo, cmds);
// key update command
blockManager.addKeyUpdateCommand(cmds, nodeinfo);
// check for balancer bandwidth update
if (nodeinfo.getBalancerBandwidth() > 0) {
cmds.add(new BalancerBandwidthCommand(nodeinfo.getBalancerBandwidth()));
// set back to 0 to indicate that datanode has been sent the new value
nodeinfo.setBalancerBandwidth(0);
}
if (slowPeerTracker != null) {
final Map<String, Double> slowPeersMap = slowPeers.getSlowPeers();
if (!slowPeersMap.isEmpty()) {
if (LOG.isDebugEnabled()) {
LOG.debug("DataNode " + nodeReg + " reported slow peers: " +
slowPeersMap);
}
for (String slowNodeId : slowPeersMap.keySet()) {
slowPeerTracker.addReport(slowNodeId, nodeReg.getIpcAddr(false));
}
}
}
if (slowDiskTracker != null) {
if (!slowDisks.getSlowDisks().isEmpty()) {
if (LOG.isDebugEnabled()) {
LOG.debug("DataNode " + nodeReg + " reported slow disks: " +
slowDisks.getSlowDisks());
}
slowDiskTracker.addSlowDiskReport(nodeReg.getIpcAddr(false), slowDisks);
}
}
if (!cmds.isEmpty()) {
return cmds.toArray(new DatanodeCommand[cmds.size()]);
}
return new DatanodeCommand[0];
}
定時輪循+limit 1000個塊洗掉的特性決定了hdfs洗掉資料并不會立即真正的執行物理洗掉，并且一次洗掉的數量也有限，所以出現誤刪操作需要立即停止HDFS，雖然有的資料在輪循中已被洗掉，所以事發后停止HDFS集群越早，被刪的資料越少，損失越小！
EditLog
EditLog記錄了hdfs操作的每一條日志記錄，包括當然包括洗掉，我們所熟知的檔案操作型別只有增、刪、改，但是在 HDFS 的領域里，遠遠不止這些操作，我們看看 EditLog 操作型別的列舉類org.apache.hadoop.hdfs.server.namenode.FSEditLogOpCodes
/**
* 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.hadoop.hdfs.server.namenode;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.hdfs.server.namenode.FSEditLogOp.*;
/**
* Op codes for edits file
*/
@InterfaceAudience.Private
@InterfaceStability.Unstable
public enum FSEditLogOpCodes {
// last op code in file
OP_ADD ((byte) 0, AddOp.class),
// deprecated operation
OP_RENAME_OLD ((byte) 1, RenameOldOp.class),
OP_DELETE ((byte) 2, DeleteOp.class),
OP_MKDIR ((byte) 3, MkdirOp.class),
OP_SET_REPLICATION ((byte) 4, SetReplicationOp.class),
@Deprecated OP_DATANODE_ADD ((byte) 5), // obsolete
@Deprecated OP_DATANODE_REMOVE((byte) 6), // obsolete
OP_SET_PERMISSIONS ((byte) 7, SetPermissionsOp.class),
OP_SET_OWNER ((byte) 8, SetOwnerOp.class),
OP_CLOSE ((byte) 9, CloseOp.class),
OP_SET_GENSTAMP_V1 ((byte) 10, SetGenstampV1Op.class),
OP_SET_NS_QUOTA ((byte) 11, SetNSQuotaOp.class), // obsolete
OP_CLEAR_NS_QUOTA ((byte) 12, ClearNSQuotaOp.class), // obsolete
OP_TIMES ((byte) 13, TimesOp.class), // set atime, mtime
OP_SET_QUOTA ((byte) 14, SetQuotaOp.class),
// filecontext rename
OP_RENAME ((byte) 15, RenameOp.class),
// concat files
OP_CONCAT_DELETE ((byte) 16, ConcatDeleteOp.class),
OP_SYMLINK ((byte) 17, SymlinkOp.class),
OP_GET_DELEGATION_TOKEN ((byte) 18, GetDelegationTokenOp.class),
OP_RENEW_DELEGATION_TOKEN ((byte) 19, RenewDelegationTokenOp.class),
OP_CANCEL_DELEGATION_TOKEN ((byte) 20, CancelDelegationTokenOp.class),
OP_UPDATE_MASTER_KEY ((byte) 21, UpdateMasterKeyOp.class),
OP_REASSIGN_LEASE ((byte) 22, ReassignLeaseOp.class),
OP_END_LOG_SEGMENT ((byte) 23, EndLogSegmentOp.class),
OP_START_LOG_SEGMENT ((byte) 24, StartLogSegmentOp.class),
OP_UPDATE_BLOCKS ((byte) 25, UpdateBlocksOp.class),
OP_CREATE_SNAPSHOT ((byte) 26, CreateSnapshotOp.class),
OP_DELETE_SNAPSHOT ((byte) 27, DeleteSnapshotOp.class),
OP_RENAME_SNAPSHOT ((byte) 28, RenameSnapshotOp.class),
OP_ALLOW_SNAPSHOT ((byte) 29, AllowSnapshotOp.class),
OP_DISALLOW_SNAPSHOT ((byte) 30, DisallowSnapshotOp.class),
OP_SET_GENSTAMP_V2 ((byte) 31, SetGenstampV2Op.class),
OP_ALLOCATE_BLOCK_ID ((byte) 32, AllocateBlockIdOp.class),
OP_ADD_BLOCK ((byte) 33, AddBlockOp.class),
OP_ADD_CACHE_DIRECTIVE ((byte) 34, AddCacheDirectiveInfoOp.class),
OP_REMOVE_CACHE_DIRECTIVE ((byte) 35, RemoveCacheDirectiveInfoOp.class),
OP_ADD_CACHE_POOL ((byte) 36, AddCachePoolOp.class),
OP_MODIFY_CACHE_POOL ((byte) 37, ModifyCachePoolOp.class),
OP_REMOVE_CACHE_POOL ((byte) 38, RemoveCachePoolOp.class),
OP_MODIFY_CACHE_DIRECTIVE ((byte) 39, ModifyCacheDirectiveInfoOp.class),
OP_SET_ACL ((byte) 40, SetAclOp.class),
OP_ROLLING_UPGRADE_START ((byte) 41, RollingUpgradeStartOp.class),
OP_ROLLING_UPGRADE_FINALIZE ((byte) 42, RollingUpgradeFinalizeOp.class),
OP_SET_XATTR ((byte) 43, SetXAttrOp.class),
OP_REMOVE_XATTR ((byte) 44, RemoveXAttrOp.class),
OP_SET_STORAGE_POLICY ((byte) 45, SetStoragePolicyOp.class),
OP_TRUNCATE ((byte) 46, TruncateOp.class),
OP_APPEND ((byte) 47, AppendOp.class),
OP_SET_QUOTA_BY_STORAGETYPE ((byte) 48, SetQuotaByStorageTypeOp.class),
OP_ADD_ERASURE_CODING_POLICY ((byte) 49, AddErasureCodingPolicyOp.class),
OP_ENABLE_ERASURE_CODING_POLICY((byte) 50, EnableErasureCodingPolicyOp.class),
OP_DISABLE_ERASURE_CODING_POLICY((byte) 51,
DisableErasureCodingPolicyOp.class),
OP_REMOVE_ERASURE_CODING_POLICY((byte) 52, RemoveErasureCodingPolicyOp.class),
// Note that the current range of the valid OP code is 0~127
OP_INVALID ((byte) -1);
private final byte opCode;
private final Class<? extends FSEditLogOp> opClass;
/**
* Constructor
*
* @param opCode byte value of constructed enum
*/
FSEditLogOpCodes(byte opCode) {
this(opCode, null);
}
FSEditLogOpCodes(byte opCode, Class<? extends FSEditLogOp> opClass) {
this.opCode = opCode;
this.opClass = opClass;
}
/**
* return the byte value of the enum
*
* @return the byte value of the enum
*/
public byte getOpCode() {
return opCode;
}
public Class<? extends FSEditLogOp> getOpClass() {
return opClass;
}
private static final FSEditLogOpCodes[] VALUES;
static {
byte max = 0;
for (FSEditLogOpCodes code : FSEditLogOpCodes.values()) {
if (code.getOpCode() > max) {
max = code.getOpCode();
}
}
VALUES = new FSEditLogOpCodes[max + 1];
for (FSEditLogOpCodes code : FSEditLogOpCodes.values()) {
if (code.getOpCode() >= 0) {
VALUES[code.getOpCode()] = code;
}
}
}
/**
* Converts byte to FSEditLogOpCodes enum value
*
* @param opCode get enum for this opCode
* @return enum with byte value of opCode
*/
public static FSEditLogOpCodes fromByte(byte opCode) {
if (opCode >= 0 && opCode < VALUES.length) {
return VALUES[opCode];
}
return opCode == -1 ? OP_INVALID : null;
}
}
總計54種操作型別！打破了人們印象中檔案只有增刪改讀的幾種操作，在hadoop的配置引數dfs.namenode.name.dir可以找到路徑
這里EditLog檔案是序列化后的二進制檔案不能直接查看，hdfs自帶了決議的命令，可以決議成xml明文格式
hdfs oev -i edits_0000000000035854978-0000000000035906741 -o edits.xml
對 hdfs 的每一個操作都會記錄一串 RECORD，RECORD 里面不同的操作包含的欄位屬性也不同，但是所有操作都具備的屬性是 OPCODE，對應上面的列舉類org.apache.hadoop.hdfs.server.namenode.FSEditLogOpCodes中的操作
hdfs元資料的加載
hdfs 啟動時，NameNode 會加載 Fsimage，Fsimage 記錄了 hdfs 現有的全量的路徑資訊，啟動程序中僅僅加載 Fsimage？這句話不完全正確！啟動的同時，還會加載未被合并成 fsimage 的EditLog，關于 fsimage 具體細節這里不展開，舉個栗子：
假設Hadoop 3分鐘checkpoint一次生成Fsimage檔案，EditLog 1分鐘生成一個檔案，下面是依次生成的檔案：
fsimage_1
editlog_1
editlog_2
editlog_3
fsimage_2
editlog_4
editlog_5
當NameNode啟動時，會加載后綴時間戳最大的那個fsimage檔案和它后面產生的editlog檔案，也就是會加載fsimage_2、editlog_4、editlog_5進NameNode記憶體，
恢復方法
假設我們執行 hdfs dfs -rmr xxx 命令的操作記錄在了 editlog_5 上面，那么，重啟 NameNode 后，我們查看hdfs無法再查看到xxx路徑，如果我們把fsimage_2刪掉，NameNode則會加載fsimage_1、editlog_1、editlog_2，此時的元資料里面xxx還未被洗掉，如果此時DataNode未物理洗掉block，則資料可以恢復，但是editlog_4、editlog_5對應的hdfs操作會丟失，有沒有更好的方法呢？
方案一：
刪掉 fsimage_2，從上一次 checkpoint 的地方也就是 fsimage_1 恢復，我們集群的實際配置，是一個小時生成一次 fsimage 檔案，也就是說，這種恢復方案會導致近一小時 hdfs 新增的檔案全部丟失，這一個小時不知道發生了多少事情，可想而知的后果是恢復之后一堆報錯，顯然不是最好的方案
方案二：
修改editlog_5，把洗掉xxx那條操作改成其它安全的操作型別，這樣重啟NameNode后，又可以看到這個路徑，
步驟：
-
關閉HDFS集群
-
決議editlog
找到洗掉操作時間點范圍內所屬的editlog檔案，決議
hdfs oev -i edits_0000000000000000336-0000000000000000414 -o edits.xml查看editlog.xml，執行洗掉操作的日志已經記錄在里面了
- 替換洗掉操作
把OP_DELETE操作替換成比較安全的操作，例如
<RECORD>
<OPCODE>OP_DELETE</OPCODE>
<DATA>
<TXID>374</TXID>
<LENGTH>0</LENGTH>
<PATH>/hbase/oldWALs/test%2C16020%2C1630456996531.1630464212306</PATH>
<TIMESTAMP>1630724840046</TIMESTAMP>
<RPC_CLIENTID>d1e2ae59-ba0c-4385-87d4-4da3d9ec019b</RPC_CLIENTID>
<RPC_CALLID>1049</RPC_CALLID>
</DATA>
</RECORD>
注意：只能改變OPCODE，其他不能修改！！！
-
反決議成editlog
# 反決議更改后的xml檔案成editlog
hdfs oev -i editlog.xml -o edits_0000000000000000336-0000000000000000414.tmp -p binary
# 重命名掉之前的editlog
mv edits_0000000000000000336-0000000000000000414 edits_0000000000000000336-0000000000000000414.bak
# 替換反決議后的editlog
mv edits_0000000000000000336-0000000000000000414.tmp edits_0000000000000000336-0000000000000000414
- 使用 scp 將修改后的 editlog 同步到其他 JournalNode 節點
- 重啟 HDFS，完成檔案恢復
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/303002.html
標籤：其他