图文精华

About云-梭伦科技»专题 › 交流区 › 技术交流 › hadoop MapReduce主程序中文乱码解决
分享

hadoop MapReduce主程序中文乱码解决

hyj 发表于 2014-4-13 15:53:04 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 0 32468
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.  */

  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）即可，如下代码： 
  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，如： 
  1. job.setOutputFormatClass(GbkOutputFormat.class);
复制代码





参考：
  1. http://semantic.iteye.com/blog/1846238
复制代码




回复

使用道具 举报

自定义广告语

没找到任何评论，期待你打破沉寂

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
hyj

超级版主

关注

  • 1689

    主题

  • 2216

    帖子

  • 469

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000