在2011年7月12日 (二) 05:48的當(dāng)前修訂版本

• 編輯 建平

前言

在本文檔中,，你將了解到如何用Java接口讀寫Hadoop分布式系統(tǒng)中的文件，以及編碼的轉(zhuǎn)換等問(wèn)題,。其中有些細(xì)節(jié),，在你不知道的時(shí)候，是非常容易出錯(cuò)的,。 這邊讀寫文件分以下三種情況：

• 1. 在非Map Reduce過(guò)程中讀寫分布式文件系統(tǒng)中的文件

比如說(shuō),，你想自己遍歷一個(gè)文件，想截?cái)嘁粋€(gè)文件,，都屬于這種方式,。一般該過(guò)程發(fā)生在run函數(shù)中，程序員處理Map Reduce產(chǎn)生的中間文件上,。

• 2. 在map（或reduce）函數(shù)中讀寫一個(gè)Record。

對(duì)于TextInputFormat,，一個(gè)Record就是一行,。我們會(huì)得到一個(gè)Text對(duì)象，作為一行,。要注意的是如果讀入的文件不是UTF-8 格式（比如GBK,因?yàn)門extInputFormat只能解碼UTF-8文件,，直接讀會(huì)產(chǎn)生亂碼），我們?nèi)绾握_轉(zhuǎn)換成Unicode,。

• 3. 在map（或reduce）函數(shù)中,，讀寫文件

比如說(shuō)，在map函數(shù)中,，你想通過(guò)讀入文件初始化一個(gè)HashMap,。

非Map Reduce過(guò)程中讀文件

主要用到FileSystem類，打開一個(gè)文件后得到FSDataInputStream,，據(jù)說(shuō)這個(gè)FSDataInputStream會(huì)對(duì)數(shù)據(jù)緩 存,，所以沒(méi)必要包裝成一個(gè)BufferedReader，但其readLine()方法是被deprecated,，所以你要想一行行的讀,，還是轉(zhuǎn)成 BufferedReader吧。在轉(zhuǎn)換的過(guò)程中,，還能指定輸入文件的編碼,，比如這邊是"UTF-8"。 該讀文件的方式無(wú)法在map或reduce方法中調(diào)用,，因?yàn)樗鼈兲幱谝粋€(gè)static類中,，無(wú)法創(chuàng)建一個(gè)JobConf,。

                JobConf confQ = new JobConf(getConf(), XXXX.class);
FileSystem fs= FileSystem.get(confQ);
FSDataInputStream fin = fs.open(new Path("filePathXX"));
BufferedReader in = null;
String line;
try {
in = new BufferedReader(new InputStreamReader(fin, "UTF-8"));
while ((line = in.readLine()) != null) {
//...
}
} finally {
if (in != null) {
in.close();
}
}

非Map Reduce過(guò)程中寫文件

該過(guò)程和讀文件對(duì)應(yīng)，注意在close之前,，調(diào)用BufferedWriter的flush()方法,，不然有數(shù)據(jù)會(huì)沒(méi)寫出。

                JobConf confQ = new JobConf(getConf(), XXXX.class);
FileSystem fs= FileSystem.get(confQ);
FSDataOutputStream fout = fs.create(new Path("要寫入的文件全路徑"));
BufferedWriter out = null;
try {
out = new BufferedWriter(new OutputStreamWriter(fout, "UTF-8"));
out.write("XXXXXX");
out.newLine();
}
out.flush();
} finally {
if (out != null) {
out.close();
}
}

Map或reduce方法中需要讀取一個(gè)文件的解決方案

對(duì)于很大的文件,，目前的方案可能是預(yù)先把這個(gè)文件部署到每個(gè)集群節(jié)點(diǎn)上,。這邊講兩種小文件的處理方法。

• 方法一：把文件打包到運(yùn)行的jar包中（可以放在根目錄下）,，然后用以下方式讀取(fileName就是在根目錄下要讀取的文件名):

                BufferedReader in = null;
try {
InputStream fstream = Thread.currentThread()
.getContextClassLoader().getResourceAsStream("fileName");
in = new BufferedReader(new InputStreamReader(new DataInputStream(
fstream), "UTF-8"));
String line;
while ((line = in.readLine()) != null) {
//...
}
} finally {
in.close();
}

• 方法二：在Map或Reduce類中,，復(fù)寫public void configure(JobConf jobIn)方法（在基類MapReduceBase中定義）

   public void configure(JobConf jobIn) {
super.configure(jobIn);
try{
FileSystem fs = null;
fs = FileSystem.get(jobIn);
FSDataInputStream in;
BufferedReader bufread;
String strLine;
String[] strList;
IpField ipField;
Path IpCityPath = new Path("/user/hadoop/dw/dim/ip_cityid.txt");
if (!fs.exists(IpCityPath))
throw new IOException("Input file not found");
if (!fs.isFile(IpCityPath))
throw new IOException("Input should be a file");
in = fs.open(IpCityPath);
bufread = new BufferedReader(new InputStreamReader(in));
while ((strLine = bufread.readLine()) != null) {
strList =strLine.split("\"");
if(strList.length < 3)
continue;
IpField nodeIp = new IpField(strList[0], strList[1], strList[2]);
CityIpLocal.add(nodeIp);
}
in.close();
}
catch(IOException e)
{
e.printStackTrace();
}
}

CityIpLocal可以是外部類的一個(gè)ArrayList對(duì)象

map方法中讀GBK文件

主要是編碼轉(zhuǎn)換的問(wèn)題，我曾經(jīng)在這個(gè)問(wèn)題上調(diào)試蠻久的,。我們輸入的文件是GBK編碼的,，map中，調(diào)用Text.toString并不會(huì)把編碼自動(dòng)轉(zhuǎn)換成Unicode,，但Java中的字符串都是當(dāng)Unicode來(lái)處理的,，我們需要手動(dòng)轉(zhuǎn)換，方式如下：

Text value;//map傳入的參數(shù)
String line = new String(value.getBytes(), 0, value.getLength(),"GBK");

這邊指定的GBK代表讀入文件的編碼,，line返回的是解碼成Unicode后的結(jié)果,。 特別注意的是，這邊需要指出value.getBytes()得到的byte數(shù)組的起始和長(zhǎng)度,，（0, value.getLength()）,。 千萬(wàn)不要這樣調(diào)用：String line = new String(value.getBytes(),"GBK");這樣得到的結(jié)果，末尾可能會(huì)有多余字串,?？磥?lái)內(nèi)部是這樣實(shí)現(xiàn)的：Text對(duì)象是被復(fù)用 的，共用一個(gè)byte數(shù)組（也可能是char數(shù)組）來(lái)存東西,，下一次只是從頭開始覆寫這個(gè)數(shù)組,，到后面如果沒(méi)寫到，那么原來(lái)的內(nèi)容還會(huì)在,。

引用一段鎮(zhèn)方的話

  如果以TextInputFormat使用文件作為輸入,，map的輸入value為Text類型，text內(nèi)部實(shí)際維護(hù)的是一個(gè)byte數(shù)組,，
從輸入文件中直接以byte的形式讀入,，不會(huì)對(duì)byte做轉(zhuǎn)義：輸入文件中的字節(jié)流直接進(jìn)入text的byte數(shù)組。
Text假設(shè)內(nèi)部編碼為utf8,，調(diào)用Text.toString,Text會(huì)把內(nèi)部byte當(dāng)作utf8處理,，如果讀入文件的實(shí)際編碼為gbk
就會(huì)產(chǎn)生亂碼；此時(shí)可以通過(guò) new String(Text.getBytes(), 0, Text.getLength(), "輸入文件實(shí)際編碼")實(shí)現(xiàn)
正確轉(zhuǎn)義。
以TextOutputFormat作為輸出時(shí),，reduce過(guò)程會(huì)強(qiáng)制以u(píng)tf8編碼輸出,。但是這一步有一個(gè)小triky，Text的輸出會(huì)
調(diào)用它的Text.getBytes方法,，而由于Text內(nèi)部以byte數(shù)組形式存儲(chǔ),，實(shí)際上可以放任意內(nèi)容。
注意：1. Hadoop的TextInput/TextOutput很系統(tǒng)編碼完全無(wú)關(guān),，是通過(guò)代碼硬性寫入為UTF8的,。
2. Text和String對(duì)數(shù)據(jù)的處理是很不一樣的

Map Reduce輸出結(jié)果為GBK文件

這個(gè)是竹莊給的解決方法中文問(wèn)題 基本上就是因?yàn)門extOutputFormat把輸出編碼寫死成UTF-8了，自己把這個(gè)改掉就可以輸出GBK了,。

保存成Excel能識(shí)別的編碼（Unicode）

有的需求,，需要用Excel來(lái)打開最后生成的文件。現(xiàn)在的方式是用Tab分割字段,，這樣就能在Excel中顯示為不同的欄了,。 困難一些的是保存的編碼形式。我研究了下,，發(fā)現(xiàn)本地的Excel文件是用Unicode編碼的,，于是在Java程序中注明了Unicode，或者UTF- 16,，結(jié)果還是不行,。再看一下，原來(lái)需要的是小端的Unicode,，而默認(rèn)UTF-16保存的是大端的,，于是改成了UTF-16LE,，指定為小端的,。結(jié)果 還是失敗，能用記事本正確打開,，卻不能用Excel打開,。后來(lái)研究了下，原來(lái)還需要指定字節(jié)序的（BOM,Byte Order Mark）,。最后寫了如下代碼搞定(這個(gè)方法把UTF-8的文件轉(zhuǎn)成了Unicode的)：

        private void convertToUnicode(FSDataInputStream convert_in,
FSDataOutputStream convert_out,String head) throws IOException {
BufferedReader in = null;
BufferedWriter out = null;
try {
in = new BufferedReader(new InputStreamReader(convert_in, "UTF-8"));
out = new BufferedWriter(new OutputStreamWriter(convert_out,
"UTF-16LE"));
out.write("\uFEFF");
out.write(head+"\n");
String line;
while ((line = in.readLine()) != null) {
out.write(line);
out.newLine();
}
out.flush();
} finally {
if(in!=null)
in.close();
if(out!=null)
out.close();
}
}

這邊還有個(gè)值得注意的現(xiàn)象,out.write("\uFEFF");我們可以查得,，Unicode小端的BOM應(yīng)該是FFFE，這邊為什么反過(guò)來(lái) 了呢,？原因在于Java的字節(jié)碼順序是大端的,，\uFEFF在內(nèi)存中表示為FF在前（低地址），F(xiàn)E在后,，寫出后即為FFFE（用UltraEditor 16進(jìn)制查看,，的確如此）。這邊的現(xiàn)象應(yīng)該和機(jī)器的大小端無(wú)關(guān)的。