一、DistributedCache簡介
DistributedCache是hadoop框架提供的一種機制,可以將job指定的文件,在job執行前,先行分發到task執行的機器上,並有相關機制對cache文件進行管理.
常見的應用場景有:
分發第三方庫(jar,so等);分發算法需要的詞典文件;分發程序運行需要的配置;分發多表數據join時小表數據簡便處理等
主要的注意事項有:
1.DistributedCache只能應用於分布式的情況,包括偽分布式,完全分布式.有些api在這2種情況下有移植性問題.
2.需要分發的文件,必須提前放到hdfs上.默認的路徑前綴是hdfs://的,不是file://
3.需要分發的文件,最好在運行期間是只讀的.
4.不建議分發較大的文件,比如壓縮文件,可能會影響task的啟動速度.
二、DistributedCache的相關配置
MRv1
| 屬性名 | 默認值 | 備注 |
| ${hadoop.tmp.dir}/mapred/local |
The local directory where MapReduce stores intermediate data files. May be a comma-separated list of directories on different devices in order to spread disk i/o. Directories that do not exist are ignored. |
|
| 10737418240(10G) | The number of bytes to allocate in each local TaskTracker directory for holding Distributed Cache data. |
|
| 10000 | The maximum number of subdirectories that should be created in any particular distributed cache store. After this many directories have been created, cache items will be expunged regardless of whether the total size threshold has been exceeded. |
|
| mapreduce.tasktracker.cache.local.keep.pct |
0.95(作用於上面2個參數) | It is the target percentage of the local distributed cache that should be kept in between garbage collection runs. In practice it will delete unused distributed cache entries in LRU order until the size of the cache is less than mapreduce.tasktracker.cache.local.keep.pct of the maximum cache size. This is a floating point value between 0.0 and 1.0. The default is 0.95. |
MRv2
新的yarn架構的代碼還沒有看過,不過從配置里可以看出相關的如下配置,本文主要基於MRv1.
yarn.nodemanager.delete.debug-delay-sec
yarn.nodemanager.local-cache.max-files-per-directory
三、DistributedCache的使用方式
1.通過配置
可以配置這三個屬性值:
mapred.cache.files,
mapred.cache.archives,
mapred.create.symlink (值設為yes 如果要建link的話)
如果要分發的文件有多個的話,要以逗號分隔(貌似在建link的時候,逗號分隔前后還不能有空格,否則會報錯)
2.使用命令行
在pipes和streaming里面可能會用到
-files Specify comma-separated files to be copied to the Map/Reduce cluster
-libjars Specify comma-separated jar files to include in the classpath
-archives Specify comma-separated archives to be unarchived on the compute machines
例如:
-files hdfs://host:fs_port/user/testfile.txt
-files hdfs://host:fs_port/user/testfile.txt#testfile
-files hdfs://host:fs_port/user/testfile1.txt,hdfs://host:fs_port/user/testfile2.txt
-archives hdfs://host:fs_port/user/testfile.jar
-archives hdfs://host:fs_port/user/testfile.tgz#tgzdir
3.代碼調用
DistributedCache.addCacheFile(URI,conf) / DistributedCache.addCacheArchive(URI,conf)
DistributedCache.setCacheFiles(URIs,conf) / DistributedCache.setCacheArchives(URIs,conf)
如果要建link,需要增加DistributedCache.createSymlink(Configuration)
獲取cache文件可以使用
getLocalCacheFiles(Configuration conf)
getLocalCacheArchives(Configuration conf)
代碼調用常常會有各樣的問題,一般我比較傾向於通過createSymlink的方式來使用,就把cache當做當前目錄的文件來操作,簡單很多.
常見的通過代碼來讀取cache文件的問題如下:
a.getLocalCacheFiles在偽分布式情況下,常常返回null.
b.getLocalCacheFiles其實是把DistributedCache中的所有文件都返回.需要自己篩選出所需的文件.archives也有類似的問題.
c.getLocalCacheFiles返回的是tt機器本地文件系統的路徑,使用的時候要注意,因為很多地方默認的都是hdfs://,可以自己加上file://來避免這個問題
4.symlink
給分發的文件,在task運行的當前工作目錄建立軟連接,在使用起來的時候會更方便.沒有上面的各種麻煩
mapred.create.symlink 需要設置為yes,不是true或Y之類哦
5.實際文件存放情況
下圖顯示的為tt機器上實際文件的狀況 (只有yarn集群的截圖)
四、DistributedCache的內部基本流程
1.每個tasktracker啟動時,都會產生一個TrackerDistributedCacheManager對象,用來管理該tt機器上所有的task的cache文件.
2.在客戶端提交job時,在JobClient內,對即將cache的文件,進行校驗
以確定文件是否存在,文件的大小,文件的修改時間,以及文件的權限是否是private or public.
3.當task在tt初始化job時,會由TrackerDistributedCacheManager產生一個TaskDistributedCacheManager對象,來管理本task的cache文件.
4.和本task相關聯的TaskDistributedCacheManager,獲取並解壓相關cache文件到本地相應目錄
如果本tt機器上已經有了本job的其他task,並已經完成了相應cache文件的獲取和解壓工作,則不會重復進行.
如果本地已經有了cache文件,則比較修改時間和hdfs上的文件是否一致,如果一致則可以使用.
5.當task結束時,會對該cache進行ref減一操作.
6.TrackerDistributedCacheManager有一個clearup線程,每隔1min會去處理那些無人使用的,目錄大小大於local.cache.size或者子目錄個數大於mapreduce.tasktracker.cache.local.numberdirectories的cache目錄.