大數(shù)據(jù)流處理批處理數(shù)據(jù)工程【免費(fèi)下載鏈接】flink項(xiàng)目地址https://gitcode.com/gh_mirrors/fli/flink點(diǎn)擊查看免費(fèi)下載Working Directory工作目錄是 Flink 為 JobManager 與 TaskManager 進(jìn)程提供的本地持久化目錄用于存放進(jìn)程重啟后可以恢復(fù)的運(yùn)行時(shí)信息其設(shè)計(jì)由 FLIP-198 引入、并由 FLIP-201 擴(kuò)展出跨進(jìn)程重啟的本地恢復(fù)能力。本文以當(dāng)前倉(cāng)庫(kù)Apache FlinkStandalone 部署文檔 working_directory.md 為主體結(jié)合flink-runtime/flink-core的源碼實(shí)現(xiàn)系統(tǒng)講解 Working Directory 的目錄結(jié)構(gòu)、配置項(xiàng)、進(jìn)程資源 ID 的確定規(guī)則、存儲(chǔ)在其中的三類制品以及如何利用它實(shí)現(xiàn) TaskManager 重啟后從本地快速恢復(fù)狀態(tài)。讀完本文你將能夠?yàn)?Standalone 集群正確配置工作目錄并搭建一套進(jìn)程重啟不丟失本地狀態(tài)的高可用實(shí)踐方案。什么是 Working DirectoryWorking Directory 是 Flink 進(jìn)程JobManager 和 TaskManager用于存放進(jìn)程重啟后可以恢復(fù)的信息的本地目錄。它的核心價(jià)值在于當(dāng)進(jìn)程以相同身份identity重啟、且仍然能夠訪問(wèn)承載該目錄的存儲(chǔ)卷時(shí)之前寫入工作目錄的數(shù)據(jù)可以繼續(xù)被讀取和使用從而避免從遠(yuǎn)程存儲(chǔ)重新拉取。在源碼中工作目錄由 WorkingDirectory.java 統(tǒng)一管理。該類的注釋明確指出Class that manages a working directory for a process/instance. When being instantiated, this class makes sure that the specified working directory exists.也就是說(shuō)WorkingDirectory在實(shí)例化時(shí)會(huì)確保目錄存在不存在則創(chuàng)建并在其內(nèi)部自動(dòng)規(guī)劃好一組固定的子目錄結(jié)構(gòu)。從構(gòu)造邏輯看WorkingDirectory.java每個(gè)工作目錄根下固定包含以下子目錄子目錄用途tmp/進(jìn)程臨時(shí)文件目錄創(chuàng)建時(shí)會(huì)先清空FileUtils.cleanDirectorylocalState/本地狀態(tài)目錄供本地恢復(fù)local recovery使用blobStorage/Blob 存儲(chǔ)目錄供 BlobServer / BlobCache 使用slotAllocationSnapshots/槽位分配快照目錄這些子目錄可以通過(guò)WorkingDirectory提供的getTmpDirectory()、getLocalStateDirectory()、getBlobStorageDirectory()、getSlotAllocationSnapshotDirectory()等方法在運(yùn)行時(shí)獲取。工作目錄的目錄結(jié)構(gòu)Flink 為兩類進(jìn)程分別規(guī)劃了工作目錄其命名規(guī)則如下JobManager 工作目錄WORKING_DIR_BASE/jm_JM_RESOURCE_IDTaskManager 工作目錄WORKING_DIR_BASE/tm_TM_RESOURCE_ID其中WORKING_DIR_BASE是工作目錄基路徑baseJM_RESOURCE_ID是 JobManager 進(jìn)程的資源 IDresource idTM_RESOURCE_ID是 TaskManager 進(jìn)程的資源 ID。也就是說(shuō)同一個(gè)基路徑下不同進(jìn)程通過(guò)jm_/tm_前綴與各自的資源 ID 區(qū)分目錄。這一規(guī)則與源碼中目錄生成邏輯完全一致在 ClusterEntrypointUtils.java 中g(shù)enerateTaskManagerWorkingDirectoryFile使用tm_ resourceId作為目錄名generateJobManagerWorkingDirectoryFile使用jm_ resourceId作為目錄名。整個(gè)生成流程generateWorkingDirectoryFile見(jiàn) ClusterEntrypointUtils.java的決策順序是若配置了進(jìn)程專屬的 working-dir 選項(xiàng)如process.jobmanager.working-dir/process.taskmanager.working-dir則直接以其為基路徑否則若配置了通用選項(xiàng)process.working-dir則以其為基路徑專屬選項(xiàng)通過(guò)withFallbackKeys回退到通用選項(xiàng)若以上均未配置則從io.tmp.dirs中隨機(jī)挑選一個(gè)臨時(shí)目錄作為基路徑對(duì)應(yīng)ConfigurationUtils.getRandomTempDirectory源碼中會(huì)記錄 DEBUG 日志 Picked ... randomly from the configured temporary directories to be used as working directory base.。最后再在該基路徑下拼上jm_resourceId或tm_resourceId得到最終的工作目錄。配置 Working Directory核心配置項(xiàng)Working Directory 相關(guān)配置在 ClusterOptions.java 中定義共三個(gè)配置項(xiàng)均屬于EXPER專家級(jí)集群配置Documentation.Sections.EXPERT_CLUSTER配置項(xiàng)作用默認(rèn)值process.working-dir所有 Flink 進(jìn)程共用的工作目錄基路徑WORKING_DIR_BASE未配置時(shí)默認(rèn)從io.tmp.dirs隨機(jī)挑選一個(gè)目錄process.jobmanager.working-dir僅 JobManager 使用的工作目錄基路徑未配置時(shí)回退到process.working-dirprocess.taskmanager.working-dir僅 TaskManager 使用的工作目錄基路徑未配置時(shí)回退到process.working-dir三點(diǎn)重要說(shuō)明必須指向本地目錄。process.working-dir的官方描述是 Local working directory for Flink processes它需要指向一個(gè)本地目錄而不是分布式文件系統(tǒng)路徑。專屬配置優(yōu)先通用配置兜底。源碼中JOB_MANAGER_PROCESS_WORKING_DIR_BASE和TASK_MANAGER_PROCESS_WORKING_DIR_BASE都通過(guò)withFallbackKeys(PROCESS_WORKING_DIR_BASE.key())聲明了回退鍵fallback key因此進(jìn)程級(jí)配置未設(shè)置時(shí)會(huì)自動(dòng)讀取process.working-dir。推薦顯式配置持久化基路徑。默認(rèn)行為從io.tmp.dirs隨機(jī)挑選意味著每次啟動(dòng)路徑都可能變化若希望進(jìn)程重啟后仍能訪問(wèn)舊的工作目錄就必須顯式配置一個(gè)穩(wěn)定的本地基路徑。進(jìn)程資源 ID 配置工作目錄名中包含的資源 ID 決定了目錄的確定性相關(guān)配置項(xiàng)為配置項(xiàng)作用默認(rèn)值源碼定義jobmanager.resource-id指定 JobManager 進(jìn)程的資源 ID未配置時(shí)為隨機(jī) UUIDJobManagerOptions.javataskmanager.resource-id指定 TaskManager 進(jìn)程的資源 ID未配置時(shí)為由 RpcAddress、RpcPort 和 6 位隨機(jī)字符串組成的隨機(jī)值TaskManagerOptions.java源碼注釋明確說(shuō)明JobManager 的jobmanager.resource-idIf not configured, the ResourceID will be generated randomly隨機(jī)生成。TaskManager 的taskmanager.resource-idIf not configured, the ResourceID will be generated with the RpcAddress:RpcPort and a 6-character random string. Notice that this option is not valid in Yarn and Native Kubernetes mode.由 RpcAddress:RpcPort 加 6 位隨機(jī)字符串組成且該選項(xiàng)在 Yarn 和 Native Kubernetes 模式下不生效。由于隨機(jī) ID 會(huì)導(dǎo)致每次重啟生成不同的工作目錄名若要實(shí)現(xiàn)跨重啟恢復(fù)就必須為進(jìn)程顯式配置確定性的資源 ID詳見(jiàn)下文跨進(jìn)程重啟的本地恢復(fù)。完整配置示例在 Standalone 部署的conf/flink-conf.yaml中可作如下配置# 進(jìn)程工作目錄基路徑本地目錄必須存在且可寫 process.working-dir: /path/to/working/dir/base # 可選JobManager / TaskManager 各自獨(dú)立的基路徑優(yōu)先級(jí)高于 process.working-dir # process.jobmanager.working-dir: /path/to/jm/working/dir/base # process.taskmanager.working-dir: /path/to/tm/working/dir/base # 可選指定確定性資源 ID跨重啟恢復(fù)必需 # jobmanager.resource-id: JobManager_1 # taskmanager.resource-id: TaskManager_1配置完成后啟動(dòng) Standalone 集群日志中會(huì)打印所使用的 Working DirectoryTaskManager 側(cè)見(jiàn) TaskManagerRunner.java 的LOG.info(Using working directory: {}, workingDirectory)JobManager 側(cè)見(jiàn) ClusterEntrypoint.java 的LOG.info(Using working directory: {}., workingDirectory)可以直接觀察目錄是否落在了預(yù)期位置。工作目錄中存儲(chǔ)的制品Flink 進(jìn)程會(huì)把以下三類制品寫入工作目錄1. Blob 存儲(chǔ)BlobServer / BlobCacheJobManager 的 BlobServer 與 TaskManager 的 BlobCache 使用工作目錄下的blobStorage/子目錄存放分布式緩存、用戶 JAR 等 Blob 數(shù)據(jù)。源碼中JobManager 與 TaskManager 啟動(dòng)時(shí)都會(huì)把workingDirectory.unwrap().getBlobStorageDirectory()傳給 Blob 服務(wù)組件見(jiàn) ClusterEntrypoint.java 與 TaskManagerRunner.java。2. 本地狀態(tài)local recovery當(dāng)state.backend.local-recovery新版鍵名為execution.state-recovery.from-local見(jiàn) StateRecoveryOptions.java開啟時(shí)狀態(tài)后端會(huì)把本地快照寫入工作目錄的localState/子目錄。該配置的官方描述強(qiáng)調(diào)This option configures local recovery for the state backend, which indicates whether to recovery from local snapshot. By default, local recovery is deactivated. Local recovery currently only covers keyed state backends (including both the EmbeddedRocksDBStateBackend and the HashMapStateBackend).即本地恢復(fù)默認(rèn)關(guān)閉且目前只覆蓋 keyed state 后端EmbeddedRocksDBStateBackend 與 HashMapStateBackend。TaskManager 側(cè)會(huì)把WorkingDirectory整體傳給狀態(tài)后端相關(guān)組件見(jiàn) TaskManagerRunner.java。3. RocksDB 工作目錄若狀態(tài)后端為 RocksDBRocksDB 自身的工作目錄同樣位于進(jìn)程工作目錄之下從而保證 RocksDB 的本地?cái)?shù)據(jù)文件在進(jìn)程重啟后仍然可被定位與復(fù)用。此外從WorkingDirectory源碼可以看到工作目錄還包含slotAllocationSnapshots/槽位分配快照子目錄供調(diào)度相關(guān)組件使用。跨進(jìn)程重啟的本地恢復(fù)工作原理Working Directory 的核心用途之一是配合本地恢復(fù)特性實(shí)現(xiàn)跨進(jìn)程重啟的狀態(tài)快速恢復(fù)FLIP-201 的設(shè)計(jì)目標(biāo)進(jìn)程重啟后Flink 可以直接從本地工作目錄讀取狀態(tài)快照無(wú)需再?gòu)倪h(yuǎn)程存儲(chǔ)恢復(fù)狀態(tài)信息從而顯著縮短恢復(fù)時(shí)間。要啟用這一能力需要同時(shí)滿足三個(gè)前提條件開啟本地恢復(fù)配置state.backend.local-recovery: trueTaskManager 使用確定性資源 ID通過(guò)taskmanager.resource-id顯式指定保證重啟前后資源 ID 一致從而工作目錄名tm_TM_RESOURCE_ID不變失敗進(jìn)程以相同工作目錄重啟重啟后的 TaskManager 必須能夠訪問(wèn)原來(lái)的工作目錄同一臺(tái)機(jī)器、同一個(gè)本地卷、以相同身份啟動(dòng)。配置示例文檔 working_directory.md 給出了最小可用配置process.working-dir: /path/to/working/dir/base state.backend.local-recovery: true taskmanager.resource-id: TaskManager_1 # important: Change for every TaskManager process注意配置中的關(guān)鍵提示每個(gè) TaskManager 進(jìn)程都必須使用不同的taskmanager.resource-id。這是因?yàn)楣ぷ髂夸浢再Y源 ID 區(qū)分如果多個(gè) TaskManager 共用一個(gè) ID它們將寫入同一個(gè)tm_ID目錄并互相干擾。生命周期管理目錄的創(chuàng)建與清理從源碼看工作目錄的創(chuàng)建與清理都遵循確定性優(yōu)先的原則創(chuàng)建WorkingDirectory.create(...)在進(jìn)程啟動(dòng)時(shí)確保目錄存在并初始化各子目錄見(jiàn) WorkingDirectory.java。JobManager 與 TaskManager 分別通過(guò)ClusterEntrypointUtils.createJobManagerWorkingDirectory/createTaskManagerWorkingDirectory完成ClusterEntrypointUtils.java。清理進(jìn)程正常結(jié)束時(shí)或工作目錄非確定性時(shí)會(huì)刪除整個(gè)工作目錄。TaskManager 側(cè)邏輯見(jiàn) TaskManagerRunner.javaif (!workingDirectory.isDeterministic() || terminationResult Result.SUCCESS) { workingDirectory.unwrap().delete(); }。JobManager 側(cè)邏輯與之對(duì)稱ClusterEntrypoint.java。這段邏輯的含義是當(dāng)資源 ID 為隨機(jī)生成工作目錄不確定isDeterministic() false時(shí)無(wú)論進(jìn)程因何退出都會(huì)清理工作目錄當(dāng)資源 ID 確定時(shí)僅在進(jìn)程正常退出Result.SUCCESS時(shí)才清理異常失敗則保留目錄——這正是跨重啟恢復(fù)得以成立的關(guān)鍵失敗重啟后目錄仍然存在本地狀態(tài)得以復(fù)用。配套測(cè)試驗(yàn)證倉(cāng)庫(kù)中配套的單元測(cè)試對(duì)上述行為有直接驗(yàn)證WorkingDirectoryTest.java驗(yàn)證WorkingDirectory創(chuàng)建時(shí)目錄結(jié)構(gòu)tmp/、localState/、blobStorage/、slotAllocationSnapshots/是否正確生成。ClusterEntrypointTest.java覆蓋工作目錄生成與生命周期相關(guān)行為。最佳實(shí)踐與注意事項(xiàng)何時(shí)必須顯式配置 Working Directory默認(rèn)情況下 Flink 會(huì)從io.tmp.dirs隨機(jī)挑選基路徑這對(duì)僅需臨時(shí)目錄的普通運(yùn)行沒(méi)有問(wèn)題。但以下場(chǎng)景必須顯式配置期望 TaskManager 失敗重啟后能本地恢復(fù)狀態(tài)配合state.backend.local-recovery需要把 Blob、本地狀態(tài)等數(shù)據(jù)放置到特定的高性能本地磁盤如 NVMe SSD而不是默認(rèn)臨時(shí)目錄需要多個(gè)進(jìn)程工作目錄相互隔離、可預(yù)期例如監(jiān)控、排障時(shí)能快速定位某個(gè)進(jìn)程的目錄。配置檢查清單檢查項(xiàng)說(shuō)明基路徑是本地目錄不能指向 HDFS、S3 等分布式存儲(chǔ)目錄需要存在且對(duì)運(yùn)行用戶可寫state.backend.local-recovery僅覆蓋 keyed stateRocksDB 與 HashMap 狀態(tài)后端支持其他狀態(tài)后端不適用每個(gè) TaskManager 使用獨(dú)立taskmanager.resource-id多個(gè) TM 共用 ID 會(huì)寫入同一目錄并互相干擾重啟保持相同身份與存儲(chǔ)卷訪問(wèn)進(jìn)程用戶、掛載的本地卷需與之前一致否則無(wú)法讀取舊目錄Yarn / Native Kubernetes 模式下taskmanager.resource-id不生效該配置項(xiàng)在上述資源提供者模式下無(wú)效請(qǐng)使用對(duì)應(yīng)模式下的資源 ID 管理機(jī)制與io.tmp.dirs的關(guān)系當(dāng)不配置任何 working-dir 選項(xiàng)時(shí)基路徑從io.tmp.dirs中隨機(jī)選取見(jiàn) ClusterEntrypointUtils.java。這意味著io.tmp.dirs中每個(gè)候選目錄都可能成為工作目錄基路徑。在生產(chǎn)環(huán)境中建議將工作目錄與臨時(shí)目錄分離管理顯式配置process.working-dir指向持久化本地卷讓io.tmp.dirs繼續(xù)承擔(dān)純粹的臨時(shí)文件職責(zé)。小結(jié)Working Directory 是 Flink 進(jìn)程本地持久化運(yùn)行時(shí)信息的核心機(jī)制它以WORKING_DIR_BASE/jm_JM_RESOURCE_ID、WORKING_DIR_BASE/tm_TM_RESOURCE_ID的規(guī)則組織目錄統(tǒng)一承載 Blob 存儲(chǔ)、本地狀態(tài)快照與 RocksDB 工作目錄并借助確定性資源 ID 失敗不清除目錄的生命周期策略支撐起跨進(jìn)程重啟的本地快速恢復(fù)。配置層面只需掌握三條主線基路徑選項(xiàng)process.working-dir及其進(jìn)程級(jí)變體、資源 ID 選項(xiàng)jobmanager.resource-id/taskmanager.resource-id以及本地恢復(fù)開關(guān)state.backend.local-recovery。理解并正確配置這三個(gè)維度即可在 Standalone 集群中實(shí)現(xiàn)進(jìn)程重啟后的本地狀態(tài)復(fù)用顯著降低故障恢復(fù)成本。贊分享大數(shù)據(jù)流處理批處理數(shù)據(jù)工程【免費(fèi)下載鏈接】flink項(xiàng)目地址https://gitcode.com/gh_mirrors/fli/flink點(diǎn)擊查看免費(fèi)下載相關(guān)推薦Flink Standalone 部署的 Working Directory 配置完全指南原理、配置項(xiàng)與跨進(jìn)程重啟的本地恢復(fù)Flink Standalone 部署的 Working Directory 配置完全指南原理、配置項(xiàng)與跨進(jìn)程重啟的本地恢復(fù) 本指南以 Flink 官方文檔大數(shù)據(jù)流處理批處理數(shù)據(jù)工程Hydra 工作目錄Working Directory自定義完全指南run/sweep 輸出目錄模式詳解Hydra 工作目錄Working Directory自定義完全指南run/sweep 輸出目錄模式詳解 導(dǎo)讀 Hydra 會(huì)自動(dòng)為每次運(yùn)行run和多開發(fā)工具后端CLIWezTerm 中 wezterm set-working-directory 命令完全指南OSC 7 工作目錄通知的原理、用法與集成實(shí)踐WezTerm 中 wezterm set working directory 命令完全指南OSC 7 工作目錄通知的原理、用法與集成實(shí)踐 導(dǎo)讀 wezter桌面應(yīng)用開發(fā)工具跨平臺(tái)創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考