【发布时间】：2018-04-02 00:34:13
【问题描述】：
我不是 Java 专家，但我了解 Java 的基础知识，并且我总是尝试深入了解 Java 代码，无论何时遇到它。
这可能是一个非常愚蠢的疑问，但我很想在我心中清楚地理解它。
我在 Java 社区发帖，因为我的怀疑只是关于 Java。
自从最近几个月我开始使用 hadoop 以来，我发现 hadoop 使用自己的类型，这些类型围绕 Java 的原始类型进行包装，以便在序列化和反序列化的基础上提高通过网络发送数据的效率。
我的困惑从这里开始，假设我们在 HDFS 中有一些数据要使用在 hadoop 代码中运行的以下 Java 代码进行处理
org.apache.hadoop.io.IntWritable;
org.apache.hadoop.io.LongWritable;
org.apache.hadoop.io.Text;
org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
public class WordCountMapper
{
extends Mapper<LongWritable,Text,Text,IntWritable>
@Override
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{
}
}
String line = value.toString();
for (String word : line.split(" ")){
if(word.length()>0){
context.write(new Text(word),new IntWritable(1));
}
在这段代码中，hadoop 的类型有 LongWritable、Text、IntWritable。
让我们选择包裹在 Java 的 String 类型周围的 Text 类型（如果我错了，请纠正我）。
我的疑问是，当我们在上面的代码中将这些参数传递给我们的方法映射时，这些参数如何与import package i.e org.apache.hadoop.io.Text;中的代码交互
下面是Text类代码
package org.apache.hadoop.io;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.CharsetEncoder;
import java.nio.charset.CodingErrorAction;
import java.nio.charset.MalformedInputException;
import java.text.CharacterIterator;
import java.text.StringCharacterIterator;
import java.util.Arrays;
import org.apache.avro.reflect.Stringable;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.classification.InterfaceStability.Stable;
@Stringable
@InterfaceAudience.Public
@InterfaceStability.Stable
public class Text
extends BinaryComparable
implements WritableComparable<BinaryComparable>
{
private static final Log LOG = LogFactory.getLog(Text.class);
private static ThreadLocal<CharsetEncoder> ENCODER_FACTORY = new ThreadLocal()
{
protected CharsetEncoder initialValue() {
return Charset.forName("UTF-8").newEncoder().onMalformedInput(CodingErrorAction.REPORT).onUnmappableCharacter(CodingErrorAction.REPORT);
}
};
private static ThreadLocal<CharsetDecoder> DECODER_FACTORY = new ThreadLocal()
{
protected CharsetDecoder initialValue() {
return Charset.forName("UTF-8").newDecoder().onMalformedInput(CodingErrorAction.REPORT).onUnmappableCharacter(CodingErrorAction.REPORT);
}
};
private static final byte[] EMPTY_BYTES = new byte[0];
private byte[] bytes;
private int length;
public Text()
{
bytes = EMPTY_BYTES;
}
public Text(String string)
{
set(string);
}
public Text(Text utf8)
{
set(utf8);
}
public Text(byte[] utf8)
{
set(utf8);
}
public byte[] getBytes()
{
return bytes;
}
public int getLength()
{
return length;
}
public int charAt(int position)
{
if (position > length) return -1;
if (position < 0) { return -1;
}
ByteBuffer bb = (ByteBuffer)ByteBuffer.wrap(bytes).position(position);
return bytesToCodePoint(bb.slice());
}
public int find(String what) {
return find(what, 0);
}
public int find(String what, int start)
{
try
{
ByteBuffer src = ByteBuffer.wrap(bytes, 0, length);
ByteBuffer tgt = encode(what);
byte b = tgt.get();
src.position(start);
while (src.hasRemaining()) {
if (b == src.get()) {
src.mark();
tgt.mark();
boolean found = true;
int pos = src.position() - 1;
while (tgt.hasRemaining()) {
if (!src.hasRemaining()) {
tgt.reset();
src.reset();
found = false;
}
else if (tgt.get() != src.get()) {
tgt.reset();
src.reset();
found = false;
}
}
if (found) return pos;
}
}
return -1;
}
catch (CharacterCodingException e) {
e.printStackTrace(); }
return -1;
}
public void set(String string)
{
try
{
ByteBuffer bb = encode(string, true);
bytes = bb.array();
length = bb.limit();
} catch (CharacterCodingException e) {
throw new RuntimeException("Should not have happened " + e.toString());
}
}
public void set(byte[] utf8)
{
set(utf8, 0, utf8.length);
}
public void set(Text other)
{
set(other.getBytes(), 0, other.getLength());
}
public void set(byte[] utf8, int start, int len)
{
setCapacity(len, false);
System.arraycopy(utf8, start, bytes, 0, len);
length = len;
}
public void append(byte[] utf8, int start, int len)
{
setCapacity(length + len, true);
System.arraycopy(utf8, start, bytes, length, len);
length += len;
}
public void clear()
{
length = 0;
}
private void setCapacity(int len, boolean keepData)
{
if ((bytes == null) || (bytes.length < len)) {
if ((bytes != null) && (keepData)) {
bytes = Arrays.copyOf(bytes, Math.max(len, length << 1));
} else {
bytes = new byte[len];
}
}
}
public String toString()
{
try
{
return decode(bytes, 0, length);
} catch (CharacterCodingException e) {
throw new RuntimeException("Should not have happened " + e.toString());
}
}
public void readFields(DataInput in)
throws IOException
{
int newLength = WritableUtils.readVInt(in);
setCapacity(newLength, false);
in.readFully(bytes, 0, newLength);
length = newLength;
}
public static void skip(DataInput in) throws IOException
{
int length = WritableUtils.readVInt(in);
WritableUtils.skipFully(in, length);
}
public void write(DataOutput out)
throws IOException
{
WritableUtils.writeVInt(out, length);
out.write(bytes, 0, length);
}
public boolean equals(Object o)
{
if ((o instanceof Text))
return super.equals(o);
return false;
}
请问，当我们运行上述 hadoop 的代码时，HDFS 中的数据会流经我们在 map 方法中提到的参数。
一旦来自 HDFS 的第一个数据集达到 Text 参数，它如何在 org.apache.hadoop.io.Text 类中流动？
我的意思是它从哪里开始（我假设它从类中的 set 方法开始，因为它具有与提到的 map 方法相同的参数，对吗？）
代码中从普通字符串类型变为Text类型在哪里？
我的第二个疑问是：当数据以 Text 类型存储时，谁来开始进行序列化？我的意思是，一旦数据到达网络上的目的地，谁调用这个 write(DataOutput out)，谁调用 readFields(DataInput in)？
它是如何工作的，我需要在哪里查看？
我希望我问的很清楚。
【问题讨论】：
标签： java hadoop serialization mapreduce deserialization