2017-10-20 108 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