hadoop解决中文输出乱码

雨点打透心脏的1/2处 2022-09-18 15:54 507阅读 0赞

**hadoop涉及输出文本的默认输出编码统一用没有BOM的UTF-8的形式**，但是对于中文的输出window系统默认的是GBK，有些格式文件例如CSV格式的文件用excel打开输出编码为没有BOM的UTF-8文件时，输出的结果为乱码，只能由UE或者记事本打开才能正常显示。因此将hadoop默认输出编码更改为GBK成为非常常见的需求。

方法一：

String line = transformText(value, "gbk");

public static String transformText(Text text, String encoding) \{  
        String value = null;  
        try \{  
        value = new String(text.getBytes(), 0, text.getLength(), encoding);  
        \}catch (UnsupportedEncodingException e) \{  
        e.printStackTrace();  
        \}  
        return value;  
    \}

方法二：

默认的情况下MR主程序中，设定输出编码的设置语句为：

Java代码   [![收藏代码][icon_star.png]][icon_star.png 1]

1.  job.setOutputFormatClass(TextOutputFormat.**class**);

Java代码   [![收藏代码][icon_star.png]][icon_star.png 1]

1.  TextOutputFormat.**class**

的代码如下：

Java代码   [![收藏代码][icon_star.png]][icon_star.png 1]

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.  \*/  
18.   
19. **package** org.apache.hadoop.mapreduce.lib.output;  
20.   
21. **import** java.io.DataOutputStream;  
22. **import** java.io.IOException;  
23. **import** java.io.UnsupportedEncodingException;  
24.   
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;  
31.   
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.\*;  
40.   
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.     \}  
57.   
58.     **protected** DataOutputStream out;  
59.     **private** **final** **byte**\[\] keyValueSeparator;  
60.   
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.     \}  
69.   
70.     **public** LineRecordWriter(DataOutputStream out) \{  
71.       **this**(out, "\\t");  
72.     \}  
73.   
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.     \}  
88.   
89.     **public** **synchronized** **void** write(K key, V value)  
90.       **throws** IOException \{  
91.   
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.     \}  
108.   
109.     **public** **synchronized**   
110.     **void** close(TaskAttemptContext context) **throws** IOException \{  
111.       out.close();  
112.     \}  
113.   \}  
114.   
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）即可，如下代码：

Java代码   [![收藏代码][icon_star.png]][icon_star.png 1]

1.  **import** java.io.DataOutputStream;  
2.  **import** java.io.IOException;  
3.  **import** java.io.UnsupportedEncodingException;  
4.    
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;  
11.   
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.\*;  
21.   
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.     \}  
37.   
38.     **protected** DataOutputStream out;  
39.     **private** **final** **byte**\[\] keyValueSeparator;  
40.   
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.     \}  
49.   
50.     **public** LineRecordWriter(DataOutputStream out) \{  
51.       **this**(out, "\\t");  
52.     \}  
53.   
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.     \}  
68.   
69.     **public** **synchronized** **void** write(K key, V value)  
70.       **throws** IOException \{  
71.   
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.     \}  
88.   
89.     **public** **synchronized**   
90.     **void** close(TaskAttemptContext context) **throws** IOException \{  
91.       out.close();  
92.     \}  
93.   \}  
94.   
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，如：

Java代码   [![收藏代码][icon_star.png]][icon_star.png 1]

1.  job.setOutputFormatClass(GbkOutputFormat.**class**);

[icon_star.png]: http://semantic.iteye.com/images/icon_star.png
[icon_star.png 1]: /images/20220828/d3e0db3e5aec4c929605eecf97a57b6a.png