hadoop 輸出中文亂碼問題

本文轉載至：

　　http://www.aboutyun.com/thread-7358-1-1.html

 

hadoop涉及輸出文本的默認輸出編碼統一用沒有BOM的UTF-8的形式，但是對於中文的輸出window系統默認的是GBK，有些格式文件例如CSV格式的文件用excel打開輸出編碼為沒有BOM的UTF-8文件時，輸出的結果為亂碼，只能由UE或者記事本打開才能正常顯示。因此將hadoop默認輸出編碼更改為GBK成為非常常見的需求。 
      默認的情況下MR主程序中，設定輸出編碼的設置語句為：

1. job.setOutputFormatClass(TextOutputFormat.class);

復制代碼

1. TextOutputFormat.class

復制代碼

的代碼如下： 

1. /**
2. * Licensed to the Apache Software Foundation (ASF) under one
3. * or more contributor license agreements.  See the NOTICE file
4. * distributed with this work for additional information
5. * regarding copyright ownership.  The ASF licenses this file
6. * to you under the Apache License, Version 2.0 (the
7. * "License"); you may not use this file except in compliance
8. * with the License.  You may obtain a copy of the License at
9. *
10. *     http://www.apache.org/licenses/LICENSE-2.0
11. *
12. * Unless required by applicable law or agreed to in writing, software
13. * distributed under the License is distributed on an "AS IS" BASIS,
14. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15. * See the License for the specific language governing permissions and
16. * limitations under the License.
17. */
19. package org.apache.hadoop.mapreduce.lib.output;
21. import java.io.DataOutputStream;
22. import java.io.IOException;
23. import java.io.UnsupportedEncodingException;
25. import org.apache.hadoop.classification.InterfaceAudience;
26. import org.apache.hadoop.classification.InterfaceStability;
27. import org.apache.hadoop.conf.Configuration;
28. import org.apache.hadoop.fs.FileSystem;
29. import org.apache.hadoop.fs.Path;
30. import org.apache.hadoop.fs.FSDataOutputStream;
32. import org.apache.hadoop.io.NullWritable;
33. import org.apache.hadoop.io.Text;
34. import org.apache.hadoop.io.compress.CompressionCodec;
35. import org.apache.hadoop.io.compress.GzipCodec;
36. import org.apache.hadoop.mapreduce.OutputFormat;
37. import org.apache.hadoop.mapreduce.RecordWriter;
38. import org.apache.hadoop.mapreduce.TaskAttemptContext;
39. import org.apache.hadoop.util.*;
41. /** An {@link OutputFormat} that writes plain text files. */
42. @InterfaceAudience.Public
43. @InterfaceStability.Stable
44. public class TextOutputFormat<K, V> extends FileOutputFormat<K, V> {
45.   public static String SEPERATOR = "mapreduce.output.textoutputformat.separator";
46.   protected static class LineRecordWriter<K, V>
47.     extends RecordWriter<K, V> {
48.     private static final String utf8 = "UTF-8";  // 將UTF-8轉換成GBK 
49.     private static final byte[] newline;
50.     static {
51.       try {
52.         newline = "\n".getBytes(utf8);
53.       } catch (UnsupportedEncodingException uee) {
54.         throw new IllegalArgumentException("can't find " + utf8 + " encoding");
55.       }
56.     }
58.     protected DataOutputStream out;
59.     private final byte[] keyValueSeparator;
61.     public LineRecordWriter(DataOutputStream out, String keyValueSeparator) {
62.       this.out = out;
63.       try {
64.         this.keyValueSeparator = keyValueSeparator.getBytes(utf8);
65.       } catch (UnsupportedEncodingException uee) {
66.         throw new IllegalArgumentException("can't find " + utf8 + " encoding");
67.       }
68.     }
70.     public LineRecordWriter(DataOutputStream out) {
71.       this(out, "\t");
72.     }
74.     /**
75.      * Write the object to the byte stream, handling Text as a special
76.      * case.
77.      * @param o the object to print
78.      * @throws IOException if the write throws, we pass it on
79.      */
80.     private void writeObject(Object o) throws IOException {
81.       if (o instanceof Text) {
82.         Text to = (Text) o;   // 將此行代碼注釋掉
83.         out.write(to.getBytes(), 0, to.getLength());  // 將此行代碼注釋掉
84.       } else { // 將此行代碼注釋掉      
85.         out.write(o.toString().getBytes(utf8));
86.       }
87.     }
89.     public synchronized void write(K key, V value)
90.       throws IOException {
92.       boolean nullKey = key == null || key instanceof NullWritable;
93.       boolean nullValue = value == null || value instanceof NullWritable;
94.       if (nullKey && nullValue) {
95.         return;
96.       }
97.       if (!nullKey) {
98.         writeObject(key);
99.       }
100.       if (!(nullKey || nullValue)) {
101.         out.write(keyValueSeparator);
102.       }
103.       if (!nullValue) {
104.         writeObject(value);
105.       }
106.       out.write(newline);
107.     }
109.     public synchronized 
110.     void close(TaskAttemptContext context) throws IOException {
111.       out.close();
112.     }
113.   }
115.   public RecordWriter<K, V> 
116.          getRecordWriter(TaskAttemptContext job
117.                          ) throws IOException, InterruptedException {
118.     Configuration conf = job.getConfiguration();
119.     boolean isCompressed = getCompressOutput(job);
120.     String keyValueSeparator= conf.get(SEPERATOR, "\t");
121.     CompressionCodec codec = null;
122.     String extension = "";
123.     if (isCompressed) {
124.       Class<? extends CompressionCodec> codecClass = 
125.         getOutputCompressorClass(job, GzipCodec.class);
126.       codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, conf);
127.       extension = codec.getDefaultExtension();
128.     }
129.     Path file = getDefaultWorkFile(job, extension);
130.     FileSystem fs = file.getFileSystem(conf);
131.     if (!isCompressed) {
132.       FSDataOutputStream fileOut = fs.create(file, false);
133.       return new LineRecordWriter<K, V>(fileOut, keyValueSeparator);
134.     } else {
135.       FSDataOutputStream fileOut = fs.create(file, false);
136.       return new LineRecordWriter<K, V>(new DataOutputStream
137.                                         (codec.createOutputStream(fileOut)),
138.                                         keyValueSeparator);
139.     }
140.   }
141. }

復制代碼

從上述代碼的第48行可以看出hadoop已經限定此輸出格式統一為UTF-8，因此為了改變hadoop的輸出代碼的文本編碼只需定義一個和TextOutputFormat相同的類GbkOutputFormat同樣繼承FileOutputFormat（注意是org.apache.hadoop.mapreduce.lib.output.FileOutputFormat）即可，如下代碼： 

1. import java.io.DataOutputStream;
2. import java.io.IOException;
3. import java.io.UnsupportedEncodingException;
5. import org.apache.hadoop.classification.InterfaceAudience;
6. import org.apache.hadoop.classification.InterfaceStability;
7. import org.apache.hadoop.conf.Configuration;
8. import org.apache.hadoop.fs.FileSystem;
9. import org.apache.hadoop.fs.Path;
10. import org.apache.hadoop.fs.FSDataOutputStream;
12. import org.apache.hadoop.io.NullWritable;
13. import org.apache.hadoop.io.Text;
14. import org.apache.hadoop.io.compress.CompressionCodec;
15. import org.apache.hadoop.io.compress.GzipCodec;
16. import org.apache.hadoop.mapreduce.OutputFormat;
17. import org.apache.hadoop.mapreduce.RecordWriter;
18. import org.apache.hadoop.mapreduce.TaskAttemptContext;
19. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
20. import org.apache.hadoop.util.*;
22. @InterfaceAudience.Public
23. @InterfaceStability.Stable
24. public class GbkOutputFormat<K, V> extends FileOutputFormat<K, V> {
25.   public static String SEPERATOR = "mapreduce.output.textoutputformat.separator";
26.   protected static class LineRecordWriter<K, V>
27.     extends RecordWriter<K, V> {
28.     private static final String utf8 = "GBK";
29.     private static final byte[] newline;
30.     static {
31.       try {
32.         newline = "\n".getBytes(utf8);
33.       } catch (UnsupportedEncodingException uee) {
34.         throw new IllegalArgumentException("can't find " + utf8 + " encoding");
35.       }
36.     }
38.     protected DataOutputStream out;
39.     private final byte[] keyValueSeparator;
41.     public LineRecordWriter(DataOutputStream out, String keyValueSeparator) {
42.       this.out = out;
43.       try {
44.         this.keyValueSeparator = keyValueSeparator.getBytes(utf8);
45.       } catch (UnsupportedEncodingException uee) {
46.         throw new IllegalArgumentException("can't find " + utf8 + " encoding");
47.       }
48.     }
50.     public LineRecordWriter(DataOutputStream out) {
51.       this(out, "\t");
52.     }
54.     /**
55.      * Write the object to the byte stream, handling Text as a special
56.      * case.
57.      * @param o the object to print
58.      * @throws IOException if the write throws, we pass it on
59.      */
60.     private void writeObject(Object o) throws IOException {
61.       if (o instanceof Text) {
62. //        Text to = (Text) o;
63. //        out.write(to.getBytes(), 0, to.getLength());
64. //      } else {
65.         out.write(o.toString().getBytes(utf8));
66.       }
67.     }
69.     public synchronized void write(K key, V value)
70.       throws IOException {
72.       boolean nullKey = key == null || key instanceof NullWritable;
73.       boolean nullValue = value == null || value instanceof NullWritable;
74.       if (nullKey && nullValue) {
75.         return;
76.       }
77.       if (!nullKey) {
78.         writeObject(key);
79.       }
80.       if (!(nullKey || nullValue)) {
81.         out.write(keyValueSeparator);
82.       }
83.       if (!nullValue) {
84.         writeObject(value);
85.       }
86.       out.write(newline);
87.     }
89.     public synchronized 
90.     void close(TaskAttemptContext context) throws IOException {
91.       out.close();
92.     }
93.   }
95.   public RecordWriter<K, V> 
96.          getRecordWriter(TaskAttemptContext job
97.                          ) throws IOException, InterruptedException {
98.     Configuration conf = job.getConfiguration();
99.     boolean isCompressed = getCompressOutput(job);
100.     String keyValueSeparator= conf.get(SEPERATOR, "\t");
101.     CompressionCodec codec = null;
102.     String extension = "";
103.     if (isCompressed) {
104.       Class<? extends CompressionCodec> codecClass = 
105.         getOutputCompressorClass(job, GzipCodec.class);
106.       codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, conf);
107.       extension = codec.getDefaultExtension();
108.     }
109.     Path file = getDefaultWorkFile(job, extension);
110.     FileSystem fs = file.getFileSystem(conf);
111.     if (!isCompressed) {
112.       FSDataOutputStream fileOut = fs.create(file, false);
113.       return new LineRecordWriter<K, V>(fileOut, keyValueSeparator);
114.     } else {
115.       FSDataOutputStream fileOut = fs.create(file, false);
116.       return new LineRecordWriter<K, V>(new DataOutputStream
117.                                         (codec.createOutputStream(fileOut)),
118.                                         keyValueSeparator);
119.     }
120.   }
121. }

復制代碼

最后將輸出編碼類型設置成GbkOutputFormat.class，如： 

1. job.setOutputFormatClass(GbkOutputFormat.class);

復制代碼

參考：

1. http://semantic.iteye.com/blog/1846238

復制代碼