转自 https://www.cnblogs.com/teagnes/p/6112019.html

其实在很久之前洒家刚刚搞hadoop的时候就遇到了这个问题，只是那个时候只知道读取hdfs上的文本文件的时候一定要是utf8编码，不然就会出现乱码，后来倒也没遇到这个问题，毕竟平时的数据都是从hive里来的，那时候也不懂这是为什么，最近又遇到了，有感于斯，从新总结一下,如何在hadoop和spark上处理读取GBK编码文件

首先来看一下为什么会出现这个问题， 下面是一个最简单的spark的wordcount程序，sc.textFile(filePath)方法从文本文件创建RDD，传入文件路径filePath，查看textFile方法， 可以看到，实际上调用了TextInputformat类来解析文本文件，熟悉hadoop的一定知道，mapreudce默认的解析文件文件的类就是TextInputformat，并返回了K V键值对

object Wordcount {
  def main(args: Array[String]) {
     val filePath = "";
     val conf = new SparkConf().setAppName("WordCountApp")
     val sc = new SparkContext(conf)
     val line = sc.textFile(filePath)
     line.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect.foreach(println)
     sc.stop
  }
}

def textFile(
   path: String,
   minPartitions: Int = defaultMinPartitions): RDD[String] = withScope {
      assertNotStopped()
      hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],
      minPartitions).map(pair => pair._2.toString).setName(path)
}

继续看TextInputFormat源码，TextInputFormat有两个作用。

一是对输入文件分片，mapreduce会为每一个分片都起动一个map任务来处理，分片的任务由TextInputFormat的父类FileInputFormat完成，这里就不做深究了， TextInputFormat中只有读取数据的方法。

二是从分片的数据，生成k v键值对也就是Recordreader ，createRecordReader方法不断的生成Recordreader对像并交给map端去处理 ，下面的代码中在delimiter.getBytes(Charsets.UTF_8)设置了字符集，很可惜这里并不是读取文件时使用的，而是指定了redcord的分割符，默认情况下是每一行生成一个record，一般情况下我们不需要使用到这个参数，只有在设置多行作为一个record输入的时候才会用到，可以通过设置参数“textinputformat.record.delimiter”来设置，那我们是不是可以在代码中指定我们的读取文件的字符集呢？

package org.apache.hadoop.mapreduce.lib.input;

import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.SplittableCompressionCodec;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import com.google.common.base.Charsets;

/** An {@link InputFormat} for plain text files.  Files are broken into lines.
 * Either linefeed or carriage-return are used to signal end of line.  Keys are
 * the position in the file, and values are the line of text.. */
@InterfaceAudience.Public
@InterfaceStability.Stable
public class TextInputFormat extends FileInputFormat<LongWritable, Text> {

  @Override
  public RecordReader<LongWritable, Text> 
    createRecordReader(InputSplit split,
                       TaskAttemptContext context) {
    String delimiter = context.getConfiguration().get(
        "textinputformat.record.delimiter");
    byte[] recordDelimiterBytes = null;
    if (null != delimiter)
      recordDelimiterBytes = delimiter.getBytes(Charsets.UTF_8);
    return new LineRecordReader(recordDelimiterBytes);
  }

  @Override
  protected boolean isSplitable(JobContext context, Path file) {
    final CompressionCodec codec =
      new CompressionCodecFactory(context.getConfiguration()).getCodec(file);
    if (null == codec) {
      return true;
    }
    return codec instanceof SplittableCompressionCodec;
  }

}

继续看LineRecordReader类，查看其中的nextKeyValue方法，该方法是具体生成k v记录时候使用的，这里有两个很意思的点，需要注意。

一是skipUtfByteOrderMark()方法，该方法处理了当文件是有bom的utf-8格式的时候，读取程序自动跳过bom，有待具体测试一下

二是如果我们读到的行跨块了怎么处理？因为hdfs是按文件的大小来切分文件的，难免一行数据被切分到两个块中去了，这里有相应的处理的逻辑，这里就不再详细说明了

public boolean nextKeyValue() throws IOException {
    if (key == null) {
      key = new LongWritable();
    }
    key.set(pos);
    if (value == null) {
      value = new Text();
    }
    int newSize = 0;
    // We always read one extra line, which lies outside the upper
    // 具体读取记录的方法split limit i.e. (end - 1)
    while (getFilePosition() <= end || in.needAdditionalRecordAfterSplit()) {
      if (pos == 0) {
        newSize = skipUtfByteOrderMark();
      } else {
        newSize = in.readLine(value, maxLineLength, maxBytesToConsume(pos));
        pos += newSize;
      }

      if ((newSize == 0) || (newSize < maxLineLength)) {
        break;
      }

      // line too long. try again
      LOG.info("Skipped line of size " + newSize + " at pos " + 
               (pos - newSize));
    }
    if (newSize == 0) {
      key = null;
      value = null;
      return false;
    } else {
      return true;
    }
  }

这里的value就是在map端获得的value，看它是怎么被赋值的，可以看到是从输入流中读取数据，这里有两种读取的方法，默认readDefaultLine的读取一行和通过自定义readCustomLine的分隔符的跨行

public int readLine(Text str, int maxLineLength,
                      int maxBytesToConsume) throws IOException {
    if (this.recordDelimiterBytes != null) {
      return readCustomLine(str, maxLineLength, maxBytesToConsume);
    } else {
      return readDefaultLine(str, maxLineLength, maxBytesToConsume);
    }
  }

默认的方式读取文件并没有用到自定义的分割符，而value获取到的还是输入流中的字节码，所以value的获得的依旧是文件的字节码，并没有做过处理，那么我们是不是可以在map端获取到的字节码按照“GBK”的方式来解码读取呢？经过测试之后发现的确是可以正常读取的

private int readDefaultLine(Text str, int maxLineLength, int maxBytesToConsume)
  throws IOException {
    /* We're reading data from in, but the head of the stream may be
     * already buffered in buffer, so we have several cases:
     * 1. No newline characters are in the buffer, so we need to copy
     *    everything and read another buffer from the stream.
     * 2. An unambiguously terminated line is in buffer, so we just
     *    copy to str.
     * 3. Ambiguously terminated line is in buffer, i.e. buffer ends
     *    in CR.  In this case we copy everything up to CR to str, but
     *    we also need to see what follows CR: if it's LF, then we
     *    need consume LF as well, so next call to readLine will read
     *    from after that.
     * We use a flag prevCharCR to signal if previous character was CR
     * and, if it happens to be at the end of the buffer, delay
     * consuming it until we have a chance to look at the char that
     * follows.
     */
    str.clear();
    int txtLength = 0; //tracks str.getLength(), as an optimization
    int newlineLength = 0; //length of terminating newline
    boolean prevCharCR = false; //true of prev char was CR
    long bytesConsumed = 0;
    do {
      int startPosn = bufferPosn; //starting from where we left off the last time
      if (bufferPosn >= bufferLength) {
        startPosn = bufferPosn = 0;
        if (prevCharCR) {
          ++bytesConsumed; //account for CR from previous read
        }
        bufferLength = fillBuffer(in, buffer, prevCharCR);
        if (bufferLength <= 0) {
          break; // EOF
        }
      }
      for (; bufferPosn < bufferLength; ++bufferPosn) { //search for newline
        if (buffer[bufferPosn] == LF) {
          newlineLength = (prevCharCR) ? 2 : 1;
          ++bufferPosn; // at next invocation proceed from following byte
          break;
        }
        if (prevCharCR) { //CR + notLF, we are at notLF
          newlineLength = 1;
          break;
        }
        prevCharCR = (buffer[bufferPosn] == CR);
      }
      int readLength = bufferPosn - startPosn;
      if (prevCharCR && newlineLength == 0) {
        --readLength; //CR at the end of the buffer
      }
      bytesConsumed += readLength;
      int appendLength = readLength - newlineLength;
      if (appendLength > maxLineLength - txtLength) {
        appendLength = maxLineLength - txtLength;
      }
      if (appendLength > 0) {
        str.append(buffer, startPosn, appendLength);
        txtLength += appendLength;
      }
    } while (newlineLength == 0 && bytesConsumed < maxBytesToConsume);

    if (bytesConsumed > Integer.MAX_VALUE) {
      throw new IOException("Too many bytes before newline: " + bytesConsumed);
    }
    return (int)bytesConsumed;
  }

解决方法：

spark读取GBK编码文件

将value的字节码按照GBK的方式读取变成字符串，运行之后能够正常显示

object GBKtoUTF8 {
 
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName(" GBK TO UTF8")
      .setMaster("local")
 
    val sc = new SparkContext(conf)
 
    val rdd = sc.hadoopFile("F:\\data\\score.txt", classOf[TextInputFormat], classOf[LongWritable], classOf[Text], 1)
      .map(p => new String(p._2.getBytes, 0, p._2.getLength, "GBK"))
      .flatMap(s => s.split(","))
      .map(x => (x, 1))
      .reduceByKey(_ + _)
      .collect
      .foreach(println)
  }
}

hadoop读取GBK编码文件

public void map(LongWritable key, Text value, Context context) {
        try {

            String line;
            line = new String(value.getBytes(), 0, value.getLength(), "GBK");//使用GBK解析字节码 ，转成String
            logger.info("gbkstr " + line);
            
            //不要使用toStirng方法来获取字符串
            //line = value.toString();    
            //logger.info("str " + line);
            
              String[] item = line.split(",");
            for (String str : item) {
                outkey = new Text(str);
                context.write(outkey, outvalue);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }