首页 > 代码库 > hadoop 输出中文乱码问题

hadoop 输出中文乱码问题

2024-08-21 02:05:55   224人阅读

本文转载至：

　　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 = http://www.mamicode.com/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 = http://www.mamicode.com/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
复制代码

hadoop 输出中文乱码问题


联系
我们