码迷，mamicode.com
首页 > 其他好文 > 详细

hadoop 输出中文乱码问题

时间：2016-11-15 16:29:42      阅读：186      评论：0      收藏：0      [点我收藏+]

标签：sso   默认   for   pass   sys   eof   统一   .lib   argument   

本文转载至：

　　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
复制代码

hadoop 输出中文乱码问题

标签：sso   默认   for   pass   sys   eof   统一   .lib   argument   

原文地址：http://www.cnblogs.com/duking1991/p/6066027.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！