最近在hadoop實(shí)際使用中有以下幾個(gè)小細(xì)節(jié)分享: i=m5M ]Ef
1 中文問題
從url中解析出中文,但hadoop中打印出來仍是亂碼?我們?cè)?jīng)以為hadoop是不支持中文的，后來經(jīng)過查看源代碼，發(fā)現(xiàn)hadoop僅僅是不支持以gbk格式輸出中文而己。
這是TextOutputFormat.class中的代碼，hadoop默認(rèn)的輸出都是繼承自FileOutputFormat來的，F(xiàn)ileOutputFormat的兩個(gè)子類一個(gè)是基于二進(jìn)制流的輸出，一個(gè)就是基于文本的輸出TextOutputFormat。
public class TextOutputFormat<K, V> extends FileOutputFormat<K, V> {
protected static class LineRecordWriter<K, V> &E{CQ#k
implements RecordWriter<K, V> {
private static final String utf8 = “UTF-8″;//這里被寫死成了utf-8 2 kP0//
private static final byte[] newline; kTC' `xv
static { :htz]
try { 0 _!')+
newline = “/n”.getBytes(utf8); Ry$zF~[
} catch (UnsupportedEncodingException uee) {
throw new IllegalArgumentException(”can’t find ” + utf8 + ” encoding”);
}
}
… k-:wM`C
public LineRecordWriter(DataOutputStream out, String keyValueSeparator) {
this.out = out;
try {
this.keyValueSeparator = keyValueSeparator.getBytes(utf8);
} catch (UnsupportedEncodingException uee) { @r.w +E=
throw new IllegalArgumentException(”can’t find ” + utf8 + ” encoding”);
}
} ab}Kt($
…
private void writeObject(Object o) throws IOException {
if (o instanceof Text) {
Text to = (Text) o;
out.write(to.getBytes(), 0, to.getLength());//這里也需要修改 q&DM*!Jq
} else { 5 O't-'
out.write(o.toString().getBytes(utf8));
}
}
… qxQuXF>:#
} |3bCq(ZR/P
可以看出hadoop默認(rèn)的輸出寫死為utf-8，因此如果decode中文正確，那么將Linux客戶端的character設(shè)為utf-8是可以看到中文的。因?yàn)閔adoop用utf-8的格式輸出了中文。
因?yàn)榇蠖鄶?shù)數(shù)據(jù)庫是用gbk來定義字段的，如果想讓hadoop用gbk格式輸出中文以兼容數(shù)據(jù)庫怎么辦？ _.{I1*6Y2
我們可以定義一個(gè)新的類： .c5)`
public class GbkOutputFormat<K, V> extends FileOutputFormat<K, V> { sTS Nu+
protected static class LineRecordWriter<K, V>
implements RecordWriter<K, V> {
//寫成gbk即可 F"ua`ercI
private static final String gbk = “gbk”;
private static final byte[] newline;
static {
try {
newline = “/n”.getBytes(gbk);
} catch (UnsupportedEncodingException uee) { @}<b42
throw new IllegalArgumentException(”can’t find ” + gbk + ” encoding”);
}
}
… SjL&/),
public LineRecordWriter(DataOutputStream out, String keyValueSeparator) { P?o|N<46
this.out = out; X-<l+WP
try { 0,]m.)ws
this.keyValueSeparator = keyValueSeparator.getBytes(gbk); Js'j}w
} catch (UnsupportedEncodingException uee) {
throw new IllegalArgumentException(”can’t find ” + gbk + ” encoding”);
}
} J|aU}Z8m
… /(&UDG$
private void writeObject(Object o) throws IOException {
if (o instanceof Text) {
// Text to = (Text) o;
// out.write(to.getBytes(), 0, to.getLength()); +A-z>T(
// } else { @h,3"2W{Ev
out.write(o.toString().getBytes(gbk));
}
} isU4D
… eL_Il.:
}
然后在mapreduce代碼中加入conf1.setOutputFormat(GbkOutputFormat.class)
即可以gbk格式輸出中文。
2 關(guān)于計(jì)算過程中的壓縮和效率的對(duì)比問題 hf//2Vl
之前曾經(jīng)介紹過對(duì)輸入文件采用壓縮可以提高部分計(jì)算效率。現(xiàn)在作更進(jìn)一步的說明。
為什么壓縮會(huì)提高計(jì)算速度？這是因?yàn)閙apreduce計(jì)算會(huì)將數(shù)據(jù)文件分散拷貝到所有datanode上，壓縮可以減少數(shù)據(jù)浪費(fèi)在帶寬上的時(shí)間，當(dāng)這些時(shí)間大于壓縮/解壓縮本身的時(shí)間時(shí)，計(jì)算速度就會(huì)提高了。
hadoop的壓縮除了將輸入文件進(jìn)行壓縮外，hadoop本身還可以在計(jì)算過程中將map輸出以及將reduce輸出進(jìn)行壓縮。這種計(jì)算當(dāng)中的壓縮又有什么樣的效果呢？
測(cè)試環(huán)境：35臺(tái)節(jié)點(diǎn)的hadoop cluster，單機(jī)2 CPU,8 core,8G內(nèi)存，redhat 2.6.9, 其中namenode和second namenode各一臺(tái)，namenode和second namenode不作datanode
輸入文件大小為2.5G不壓縮，records約為3600萬條。mapreduce程序分為兩個(gè)job: ;R]~9Aan
job1:map將record按user字段作key拆分，reduce中作外連接。這樣最后reduce輸出為87億records，大小540G
job2:map讀入這87億條數(shù)據(jù)并輸出，reduce進(jìn)行簡(jiǎn)單統(tǒng)計(jì)，最后的records為2.5億條，大小16G
計(jì)算耗時(shí)54min
僅對(duì)第二個(gè)階段的map作壓縮(第一個(gè)階段的map輸出并不大，沒有壓縮的必要)，測(cè)試結(jié)果：計(jì)算耗時(shí)39min
可見時(shí)間上節(jié)約了15min，注意以下參數(shù)的不同。 U&W/Nj
不壓縮時(shí):
Local bytes read=1923047905109 :3[;9xCHj
Local bytes written=1685607947227 "j8`)XXa(
壓縮時(shí)： /U>|^$4 #5
Local bytes read=770579526349 |RL/2j|
Local bytes written=245469534966
本地讀寫的的數(shù)量大大降低了
至于對(duì)reduce輸出的壓縮，很遺憾經(jīng)過測(cè)試基本沒有提高速度的效果。可能是因?yàn)榈谝粋€(gè)job的輸出大多數(shù)是在本地機(jī)上進(jìn)行map，不經(jīng)過網(wǎng)絡(luò)傳輸?shù)脑颉?nbsp;
附：對(duì)map輸出進(jìn)行壓縮，只需要添加 jobConf.setMapOutputCompressorClass(DefaultCodec.class)
3 關(guān)于reduce的數(shù)量設(shè)置問題
reduce數(shù)量究竟多少是適合的。目前測(cè)試認(rèn)為reduce數(shù)量約等于cluster中datanode的總cores的一半比較合適，比如 cluster中有32臺(tái)datanode,每臺(tái)8 core，那么reduce設(shè)置為128速度最快。因?yàn)槊颗_(tái)機(jī)器8 core，4個(gè)作map,4個(gè)作reduce計(jì)算，正好合適。 u/(>a
附小測(cè)試：對(duì)同一個(gè)程序 j&[u $P*K
reduce num=32,reduce time = 6 min
reduce num=128, reduce time = 2 min
reduce num=320, reduce time = 5min
4某次正常運(yùn)行mapreduce實(shí)例時(shí),拋出錯(cuò)誤
java.io.IOException: All datanodes xxx.xxx.xxx.xxx:xxx are bad. Aborting…
at org.apache.hadoop.dfs.DFSClient$DFSOutputStream.processDatanodeError(DFSClient.java:2158)
at org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access$1400(DFSClient.java:1735)
at org.apache.hadoop.dfs.DFSClient$DFSOutputStream$DataStreamer.run(DFSClient.java:1889)
java.io.IOException: Could not get block locations. Aborting…
at org.apache.hadoop.dfs.DFSClient$DFSOutputStream.processDatanodeError(DFSClient.java:2143)
at org.apache.hadoop.dfs.DFSClient$DFSOutputStream.access$1400(DFSClient.java:1735)
at org.apache.hadoop.dfs.DFSClient$DFSOutputStream$DataStreamer.run(DFSClient.java:1889)
經(jīng)查明，問題原因是linux機(jī)器打開了過多的文件導(dǎo)致。用命令ulimit -n可以發(fā)現(xiàn)linux默認(rèn)的文件打開數(shù)目為1024，修改/ect/security/limit.conf，增加hadoop soft 65535
再重新運(yùn)行程序（最好所有的datanode都修改），問題解決
P.S:據(jù)說hadoop dfs不能管理總數(shù)超過100M個(gè)文件，有待查證
5 運(yùn)行一段時(shí)間后hadoop不能stop-all.sh的問題，顯示報(bào)錯(cuò)
no tasktracker to stop ，no datanode to stop
問題的原因是hadoop在stop的時(shí)候依據(jù)的是datanode上的mapred和dfs進(jìn)程號(hào)。而默認(rèn)的進(jìn)程號(hào)保存在/tmp下，linux 默認(rèn)會(huì)每隔一段時(shí)間（一般是一個(gè)月或者7天左右）去刪除這個(gè)目錄下的文件。因此刪掉hadoop-hadoop-jobtracker.pid和 hadoop-hadoop-namenode.pid兩個(gè)文件后，namenode自然就找不到datanode上的這兩個(gè)進(jìn)程了。
在配置文件中的export HADOOP_PID_DIR可以解決這個(gè)問題