2. 程式人生 > >hadoop之MR位元組碼轉換

hadoop之MR位元組碼轉換

阿新 • • 發佈：2019-02-04

大家如果看過hadoop的文字檔案輸入字符集格式，就知道在TextOutputFormat原始碼中寫死了輸出位元組碼格式是UTF-8，原始碼如下

  
import java.io.DataOutputStream;  
import java.io.IOException;  
import java.io.UnsupportedEncodingException;  
  
import org.apache.hadoop.conf.Configuration;  
import org.apache.hadoop.fs.FileSystem;  
import org.apache.hadoop.fs.Path;  
import org.apache.hadoop.fs.FSDataOutputStream;  
  
import org.apache.hadoop.io.NullWritable;  
import org.apache.hadoop.io.Text;  
import org.apache.hadoop.io.compress.CompressionCodec;  
import org.apache.hadoop.io.compress.GzipCodec;  
import org.apache.hadoop.mapreduce.OutputFormat;  
import org.apache.hadoop.mapreduce.RecordWriter;  
import org.apache.hadoop.mapreduce.TaskAttemptContext;  
import org.apache.hadoop.util.*;  
  
/** An {@link OutputFormat} that writes plain text files. */  
public class TextOutputFormat<K, V> extends FileOutputFormat<K, V> {//TextInputFormat是預設的輸出檔案格式   
  protected static class LineRecordWriter<K, V>//預設   
    extends RecordWriter<K, V> {  
    private static final String utf8 = "UTF-8"; //這個地方寫死了輸出字符集是UTF-8 
    private static final byte[] newline;//行結束符？   
    static {  
      try {  
        newline = "\n".getBytes(utf8);  
      } catch (UnsupportedEncodingException uee) {  
        throw new IllegalArgumentException("can't find " + utf8 + " encoding");  
      }  
    }  
  
    protected DataOutputStream out;  
    private final byte[] keyValueSeparator;//key和value的分隔符，預設的好像是Tab   
  
    public LineRecordWriter(DataOutputStream out, String keyValueSeparator) {//建構函式，初始化輸出流及分隔符    
      this.out = out;  
      try {  
        this.keyValueSeparator = keyValueSeparator.getBytes(utf8);  
      } catch (UnsupportedEncodingException uee) {  
        throw new IllegalArgumentException("can't find " + utf8 + " encoding");  
      }  
    }  
  
    public LineRecordWriter(DataOutputStream out) {//預設的分隔符   
      this(out, "\t");  
    }  
  
    /** 
     * Write the object to the byte stream, handling Text as a special輸出流是byte格式的 
     * case. 
     * @param o the object to print是要輸出的物件 
     * @throws IOException if the write throws, we pass it on 
     */  
    private void writeObject(Object o) throws IOException {//應該是一行一行的寫 key keyValueSeparator value \n   
      if (o instanceof Text) {//如果o是Text的例項   
        Text to = (Text) o;  
        out.write(to.getBytes(), 0, to.getLength());//寫出   
      } else {  
        out.write(o.toString().getBytes(utf8));  
      }  
    }  
  
    public synchronized void write(K key, V value)//給寫執行緒加鎖，寫是互斥行為   
      throws IOException {  
<span style="white-space:pre">    </span>//下面是為了判斷key和value是否為空值   
      boolean nullKey = key == null || key instanceof NullWritable;//這語句太牛了   
      boolean nullValue = value == null || value instanceof NullWritable;  
      if (nullKey && nullValue) {//   
        return;  
      }  
      if (!nullKey) {  
        writeObject(key);  
      }  
      if (!(nullKey || nullValue)) {  
        out.write(keyValueSeparator);  
      }  
      if (!nullValue) {  
        writeObject(value);  
      }  
      out.write(newline);  
    }  
  
    public synchronized   
    void close(TaskAttemptContext context) throws IOException {  
      out.close();  
    }  
  }  
  
  public RecordWriter<K, V>    getRecordWriter(TaskAttemptContext job//獲得writer例項   
                         ) throws IOException, InterruptedException {  
    Configuration conf = job.getConfiguration();  
    boolean isCompressed = getCompressOutput(job);//   
    String keyValueSeparator= conf.get("mapred.textoutputformat.separator",  
                                       "\t");  
    CompressionCodec codec = null;//壓縮格式 還是？   
    String extension = "";  
    if (isCompressed) {  
      Class<? extends CompressionCodec> codecClass =   
        getOutputCompressorClass(job, GzipCodec.class);  
      codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, conf);  
      extension = codec.getDefaultExtension();  
    }  
    Path file = getDefaultWorkFile(job, extension);//這個是獲取預設的檔案路徑及名稱，在FileOutput中有對其的實現   
    FileSystem fs = file.getFileSystem(conf);  
    if (!isCompressed) {  
      FSDataOutputStream fileOut = fs.create(file, false);  
      return new LineRecordWriter<K, V>(fileOut, keyValueSeparator);  
    } else {  
      FSDataOutputStream fileOut = fs.create(file, false);  
      return new LineRecordWriter<K, V>(new DataOutputStream  
                                        (codec.createOutputStream(fileOut)),  
                                        keyValueSeparator);  
    }  
  }  
}
但是在生產環境中，輸入輸出字符集格式總是不一定會是utf-8格式，有可能處理之後的文字要求輸出格式是GBK、BIG5等之類，作為下一個程式的輸入格式，尤其是銀行業，日誌格式一般都是GBK，指定輸出格式是GBK替換上面的原始碼中的UTF-8即可，但是字符集編碼格式那麼多，做一個大資料的平臺產品，面向的就是全世界的客戶，這樣去指定輸出格式沒有那麼自動化，受眾也是極窄的。如果是我們能夠在MR程式的設定指定字符集那就完美契合生產環境中複雜的需求。故而修改原始碼如下： 
package com.huateng.hadoop.mapred.transcoding.format;

import java.io.DataOutputStream;
import java.io.IOException;
import java.io.UnsupportedEncodingException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.ReflectionUtils;

public class EncodingOutputFormat<K, V> extends FileOutputFormat<K, V>{
	  
	  public static String SEPERATOR = "mapreduce.output.textoutputformat.separator";
	  protected static class LineRecordWriter<K, V>
	    extends RecordWriter<K, V>  {
	    private   String charset;//手動指定的引數,在這裡我們可以任意去指定一個輸出的字符集
	    private   byte[] newline;
	   
	    protected DataOutputStream out;
	    private final byte[] keyValueSeparator;
//傳入的引數，在下面寫進輸出行記錄，那麼指定的字符集引數從哪裡傳進去了？必須在MR執行的時候能夠起作用。所以我們就必須在構造器中賦予該引數
	    public LineRecordWriter(DataOutputStream out, String keyValueSeparator,String dsc_charset) {
	      this.out = out;
	      charset=dsc_charset; 
	      try {
	    	  	newline = "\n".getBytes(charset);
	    	  	this.keyValueSeparator = keyValueSeparator.getBytes(charset);
	      } catch (UnsupportedEncodingException uee) {
	    	  	throw new IllegalArgumentException("can't find " + charset + " encoding");
	      }
	    }
	    /**
	     * Write the object to the byte stream, handling Text as a special case.
	     * @param o the object to print
	     * @throws IOException if the write throws, we pass it on
	     */
	    private void writeObject(Object o) throws IOException {
	      if (o instanceof Text) {
//	        	Text to = (Text) o;
//	        	out.write(to.getBytes(), 0, to.getLength());
//	      } else {
	    	  	out.write(o.toString().getBytes(charset));
	      }
	    }

	    public synchronized void write(K key, V value)
	    		throws IOException {

	      boolean nullKey = key == null || key instanceof NullWritable;
	      boolean nullValue = value == null || value instanceof NullWritable;
	      if (nullKey && nullValue) {
	        return;
	      }
	      if (!nullKey) {
	        writeObject(key);
	      }
	      if (!(nullKey || nullValue)) {
	        out.write(keyValueSeparator);
	      }
	      if (!nullValue) {
	        writeObject(value);
	      }
	      out.write("\n".getBytes());
	    }

	    public synchronized 
	    		void close(TaskAttemptContext context) throws IOException {
	      out.close();
	    }
	  }

	  public RecordWriter<K, V> 
	         getRecordWriter(
	        		 TaskAttemptContext job
	                         ) throws IOException, InterruptedException {
	    Configuration conf = job.getConfiguration();
	    String dst_charset = job.getConfiguration().get("ark.dsccodec");
//	    compress壓縮
 
//引數就是從這裡傳進來的，我們得到MR的job的引數，進行指定
	    boolean isCompressed = getCompressOutput(job);
	    String keyValueSeparator= conf.get(SEPERATOR, "\t");
	    CompressionCodec codec = null;
	    String extension = "";
	    if (isCompressed) {
	      Class<? extends CompressionCodec> codecClass = 
	        getOutputCompressorClass(job, GzipCodec.class);
	      codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, conf);
	      extension = codec.getDefaultExtension();
	    }
//	    setOutputName(job,"transform");
	    Path file = getDefaultWorkFile(job, extension);
	    FileSystem fs = file.getFileSystem(conf);
	    if (!isCompressed) {
	      FSDataOutputStream fileOut = fs.create(file, false);
	      return new LineRecordWriter<K, V>(fileOut, keyValueSeparator,dst_charset);
	    } else {
	      FSDataOutputStream fileOut = fs.create(file, false);
	      return new LineRecordWriter<K, V>(new DataOutputStream(codec.createOutputStream(fileOut)),
	                                				keyValueSeparator,dst_charset);
	    }
	  }
	}

 
 private   String charset;//手動指定的引數,在這裡我們可以任意去指定一個輸出的字符集
/傳入的引數，在下面寫進輸出行記錄，那麼指定的字符集引數從哪裡傳進去了？必須在MR執行的時候能夠起作用。所以我們就必須在構造器中賦予該引數
	    public LineRecordWriter(DataOutputStream out, String keyValueSeparator,String dsc_charset) {}
<pre name="code" class="html">protected static class LineRecordWriter<K, V>
	    extends RecordWriter<K, V>
在LineRecordWriter的父類RecordWriter中得到指定的字符集 
<pre name="code" class="html">public RecordWriter<K, V> 
	         getRecordWriter(
	        		 TaskAttemptContext job
	                         ) throws IOException, InterruptedException {
	    Configuration conf = job.getConfiguration();
	    String dst_charset = job.getConfiguration().get("ark.dsccodec");

//引數就是從這裡傳進來的，我們得到MR的job的引數，進行指定
} 
<pre name="code" class="html">job.getConfiguration().get("ark.dsccodec")這個設定Configuration則來源於建立Job例項的進行指定，至此完成。本人生產環境（華為FI叢集）親測，JDK支援的字符集都可
轉換。具體main方法設定引數如下：
<pre name="code" class="html">
import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import com.google.common.base.Preconditions;
import com.huateng.hadoop.mapred.MapRedAdapter;
import com.huateng.hadoop.mapred.transcoding.format.EncodingOutputFormat;
//import com.huateng.hadoop.mapred.transcoding.format.GB2312OutputFormat;
//import com.huateng.hadoop.mapred.transcoding.format.GBKOutputFormat;
import com.huateng.hdfs.common.HDFSClient;
import com.huateng.util.common.StringUtils;

/*
 * @author canMao
 */
public class TranscodingJob 
{
	String other_code=null;
	private Job internalJob;
	public TranscodingJob(String in_path,String src_charset,
			String out_path,String dst_charset)throws Exception{
		Preconditions.checkArgument(!StringUtils.hasNullOrEmpty(new String[]{src_charset, dst_charset})
				," source_encoding and destination_encoding is null at least one");
		Job job = MapRedAdapter.createJob();
		job.getConfiguration().set("ark.codec", src_charset);
		job.getConfiguration().set("ark.dsccodec", dst_charset);
		job.setJarByClass(TranscodingJob.class);
		job.setMapperClass(TranscodingMapper.class);
		job.setNumReduceTasks(0);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);
		if (dst_charset.equals("UTF-8")) {
			job.setOutputFormatClass(TextOutputFormat.class);
		}else{
			job.setOutputFormatClass(EncodingOutputFormat.class);
		}
		FileInputFormat.setInputPaths(job, new Path(in_path));
		if (HDFSClient.getFileSystem().exists(new Path(out_path))) {
			HDFSClient.getFileSystem().delete(new Path(out_path),true);
		}
		FileOutputFormat.setOutputPath(job, new Path(out_path));
		internalJob = job;
	}
	
	public boolean submit() throws ClassNotFoundException, IOException, InterruptedException
	{
//		float progress=0.0f;
//		internalJob.submit();
//		while(true){
//			internalJob.mapProgress();
//		}
		
		return internalJob.waitForCompletion(true);
//		internalJob.submit();
	}
	
	
}






相關推薦

hadoop之MR位元組碼轉換

大家如果看過hadoop的文字檔案輸入字符集格式，就知道在TextOutputFormat原始碼中寫死了輸出位元組碼格式是UTF-8，原始碼如下 import java.io.DataOutputStream; import java.io.IOException

Javassist之使用位元組碼在執行時生成新的類 01

介紹 　　Javassist是一個開源的分析、編輯和建立Java位元組碼的類庫。是由東京工業大學的數學和計算機科學系的 Shigeru Chiba （千葉 滋）所建立的。它已加入了開放原始碼JBoss 應用伺服器專案，通過使用Javassist對位元組碼操作為JBoss實現動態"AOP"框架。

Instrumentation與ClassFileTransformer--位元組碼轉換工具

一個代理實現ClassFileTransformer介面用於改變執行時的位元組碼（class File），這個改變發生在jvm載入這個類之前。對所有的類載入器有效。 class File這個術語定義於虛擬機器規範3.1，指的是位元組碼的byte陣列，而不是檔案系統中的class檔案。

動態生成Java位元組碼之java位元組碼框架ASM的學習

一、什麼是ASM 　　ASM是一個java位元組碼操縱框架，它能被用來動態生成類或者增強既有類的功能。ASM 可以直接產生二進位制 class 檔案，也可以在類被載入入 Java 虛擬機器之前動態改變類行為。Java class 被儲存在嚴格格式定義的 .class

從零寫一個編譯器（十一）：程式碼生成之Java位元組碼基礎

專案的完整程式碼在 C2j-Compiler 前言 第十一篇，終於要進入程式碼生成部分了，但是但是在此之前，因為我們要做的是C語言到位元組碼的編譯，所以自然要了解一些位元組碼，但是由於C語言比較簡單，所以只需要瞭解一些位元組碼基礎 JVM的基本機制 JVM有一個執行環境叫做stack frame 這個

hadoop官網 閱讀之 MR Support for YARN Shared Cache

Overview MapReduce 支援yarn共享快取，允許mapreduce利用額外的資源快取。 他儲存了作業提交客戶端和yarn叢集之間的網路頻寬。這會節約reduce工作的提交時間和所有工作的執行時間。 Enabling/Disabling the shared cache

從位元組碼看java型別轉換 ( 深入理解 (T[]) new Object[size] )

　　我們都知道，java中對型別的檢查是很嚴格的，所以我們平操作時，也往往很小心。 　　如題： (T[]) new Object[size]，這種寫法是一般我們是不會幹的！但是有點經驗的同學，還是會遇到這樣寫的。那麼，今天咱們就來看看，像這樣的寫法對不對，也順便深入理解java的型別轉換機制吧！

Python 原始碼分析之位元組碼之基本操作

本文基於 Python 3.6.4 編譯器生成位元組碼，你可通過如下程式碼片段得到 python 原始碼對應的位元組碼 #!/usr/bin/env python # encoding: utf-8 import sys import dis filename=sys.argv

網路通訊之 位元組序轉換原理與網路位元組序、大端和小端模式

     原因如下：網路協議規定接收到得第一個位元組是高位元組，存放到低地址，所以傳送時會首先去低地址取資料的高位元組。小端模式的多位元組資料在存放時，低地址存放的是低位元組，而被髮送方網路協議函式傳送時會首先去低地址取資料（想要取高位元組，真正取得是低位元組），接收方網路協議函式接收時會將接收到的第一個位元

JVM深度學習系列之位元組碼檔案學習(二)

1. 首先javac編譯java檔案  javac aa.java 2. 使用javap 反編譯class檔案得到位元組碼檔案   javap -v aa.class   Javap 相關idea外掛   jclasslib

Hadoop進階之MR中MapTask並行度決定機制及切片機制

前言 maptask的並行度決定map階段的任務處理併發度，進而影響到整個job的處理速度 那麼，mapTask並行例項是否越多越好呢？其並行度又是如何決定呢？ 1 mapTask並行度的決定機制 一個job的mapTask並行度由客戶端在

JVM Class位元組碼之三-使用BCEL改變類屬性

使用BCEL動態改變Class內容 之前對Class檔案中的常量池，Method的位元組碼指令進行了說明。JVM Class詳解之一JVM Class詳解之二 Method位元組碼指令 現在我們開始實際動手，使用BCEL改變位元組碼指令，對Class檔案進行功能擴充。

【深入Java虛擬機器】之七：深入JVM位元組碼執行引擎

我們都知道，在當前的Java中（1.0）之後，編譯器講原始碼轉成位元組碼，那麼位元組碼如何被執行的呢？這就涉及到了JVM的位元組碼執行引擎，執行引擎負責具體的程式碼呼叫及執行過程。就目前而言，所有的執行引擎的基本一致： 輸入：位元組碼檔案 處理：位元組碼解析 輸出：執

JVM系列之七：位元組碼執行

Javap– class檔案反彙編工具– javap –verbose Calcjava 檔案public class Calc {public int calc() {int a = 500;int b = 200;int c = 50;return (a + b) / c

Java學習之動態編譯--位元組碼操作--javassist類庫

一、位元組碼操作 1.Java動態性的兩種常見實現方式： 位元組碼操作反射 2.執行時操作位元組碼可以實現如下功能： 動態生成新的類動態改變某個類的結構(新增/刪除/修改 新的屬性/方法) 3.優勢： 比反射開銷小，效能高Javaasist效能高於反射，低於ASM 二、常見

深入理解Java虛擬機器之虛擬機器位元組碼執行引擎

執行引擎是java虛擬機器最核心的組成部分之一。 物理機的執行引擎是建立在處理器、硬體、指令集和作業系統層面上的，而虛擬機器的執行引擎是由自己實現的，可以自行制定指令集與執行引擎的結構體系，並且能夠執行那些硬體不直接支援的指令集格式。 執行引擎在執行Java

黑馬程式設計師——java的IO之字元流，位元組流,轉換流

------- android培訓、java培訓、期待與您交流！ ---------- 前言：通過觀看畢向東老師的java基礎視訊，查漏補缺，將一些自己掌握的還不牢固的知識寫出來，希望和大家交流分享。 1.IO概述 1、相關概念：IO：即Input和Output的縮寫。

Base64位元組碼和Bitmap之間的轉換

Bitmap轉Base64編碼字串 public String bitmaptoString(Bitmap bitmap) { // 將Bitmap轉換成字串 String string = null;

JVM之位元組碼指令簡介

java虛擬機器的指令由一個位元組長度的、代表著某種特定操作含義的數字以及跟隨其後的零至多個代表此操作所需的引數而構成的。 位元組碼與資料型別 載入和儲存指令： 將資料在棧幀中的區域性變量表和運算元棧

Android AOP之位元組碼插樁

出自：http://www.jianshu.com/p/c202853059b4 背景   本篇文章基於《網易樂得無埋點資料收集SDK》總結而成，關於網易樂得無埋點資料採集SDK的功能介紹以及技術總結後續會有文章進行闡述，本篇單講SDK中用到的Android端AOP