2017-10-20 153 views
0

我不是一个Java专家,但我知道Java的基础知识,并且总是试图在遇到任何问题时深入理解Java代码。 这可能是一个非常愚蠢的疑问,但很想清楚地理解我的想法。
我在Java社区发布,因为我的疑问只是关于Java。包装类型如何在Hadoop中工作?

自从最近几个月我和hadoop一起工作后,发现hadoop使用自己的类型,这些类型被封装在Java的原始类型中,以便在序列化和反序列化的基础上提高跨网络发送数据的效率。

我的困惑就从这里开始,可以说,我们在HDFS一些数据来使用下面的Java代码运行在Hadoop的代码

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,文本,IntWritable。
让我们拿起围绕Java的字符串类型的文本类型(纠正我,如果我错了)。
我在这里的疑问是,当我们在上面的代码,这些参数是如何获得与在import package i.e org.apache.hadoop.io.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流中的数据跨越我们在地图方法中提到的参数。
一旦来自HDFS的第一个数据集触及Text参数,它是如何在org.apache.hadoop.io.Text类中流动的呢?
我的意思是它从哪里开始(我假设它是从类中的set方法开始的,因为它具有与提到的map方法相同的参数,我正确吗?)
它从何处从普通字符串类型更改为Text键入代码?

我的第二个疑问是:当数据存储在文本类型中,然后谁踢它开始做serilzation?我的意思是说谁将这个写入(DataOutput输出),并且在数据到达网络上的目的地后调用readFields(DataInput in)?
它是如何工作的,我需要看什么?

我希望我所问的很清楚。

回答

0

与所有网络或磁盘操作一样,所有内容都以字节形式传输。 Text类将字节反序列化为UTF-8。可写入决定数据如何表示,可比数据决定数据的排序方式。

作业中设置的InputFormat决定了映射或减少任务的可写入内容。

一种InputSplit确定如何分割并读取的原始字节流进入Writables

一个地图任务对每个InputSplit

开始参见https://hadoop.apache.org/docs/stable/hadoop-mapreduce-client/hadoop-mapreduce-client-core/MapReduceTutorial.html

相关问题