Spark程序中一個(gè)Job的觸發(fā)是通過一個(gè)Action算子，比如count(), saveAsTextFile()等

創(chuàng)新互聯(lián)公司，是成都地區(qū)的互聯(lián)網(wǎng)解決方案提供商，用心服務(wù)為企業(yè)提供網(wǎng)站建設(shè)、重慶APP開發(fā)、成都微信小程序、系統(tǒng)按需定制和微信代運(yùn)營服務(wù)。經(jīng)過數(shù)十余年的沉淀與積累，沉淀的是技術(shù)和服務(wù)，讓客戶少走彎路，踏實(shí)做事，誠實(shí)做人，用情服務(wù)，致力做一個(gè)負(fù)責(zé)任、受尊敬的企業(yè)。對客戶負(fù)責(zé)，就是對自己負(fù)責(zé)，對企業(yè)負(fù)責(zé)。

在這次Spark優(yōu)化測試中，從Hive中讀取數(shù)據(jù)，將其另外保存四份，其中兩個(gè)Job采用串行方式，另外兩個(gè)Job采用并行方式。將任務(wù)提交到Y(jié)arn中執(zhí)行。能夠明顯看出串行與兵線處理的性能。

每個(gè)Job執(zhí)行時(shí)間：

四個(gè)Job都是自執(zhí)行相同操作，Job0,Job1一組采用串行方式，Job2,Job3采用并行方式。

Job0,Job1串行方式耗時(shí)等于兩個(gè)Job耗時(shí)之和 49s+39s=88s

Job2,Job3并行方式耗時(shí)等于最先開始和最后結(jié)束時(shí)間只差17:02:12-17:01:15=57s

代碼：

package com.cn.ctripotb;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.hive.HiveContext;


import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;

/**
 * Created by Administrator on 2016/9/12.
 */
public class HotelTest {
    static ResourceBundle rb = ResourceBundle.getBundle("filepath");
    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
                .setAppName("MultiJobWithThread")
                .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");

        JavaSparkContext sc = new JavaSparkContext(conf);
        HiveContext hiveContext = new HiveContext(sc.sc());  //測試真實(shí)數(shù)據(jù)時(shí)要把這里放開

        final DataFrame df = getHotelInfo(hiveContext);
        //沒有多線程處理的情況，連續(xù)執(zhí)行兩個(gè)Action操作，生成兩個(gè)Job
        df.rdd().saveAsTextFile(rb.getString("hdfspath") + "/file1",com.hadoop.compression.lzo.LzopCodec.class);
        df.rdd().saveAsTextFile(rb.getString("hdfspath") + "/file2",com.hadoop.compression.lzo.LzopCodec.class);

        //用Executor實(shí)現(xiàn)多線程方式處理Job
        java.util.concurrent.ExecutorService executorService = Executors.newFixedThreadPool(2);
        executorService.submit(new Callable() {
            @Override
            public Void call(){
                df.rdd().saveAsTextFile(rb.getString("hdfspath") + "/file3",com.hadoop.compression.lzo.LzopCodec.class);
                return null;
            }
        });
        executorService.submit(new Callable() {
            @Override
            public Void call(){
                df.rdd().saveAsTextFile(rb.getString("hdfspath") + "/file4",com.hadoop.compression.lzo.LzopCodec.class);
                return null;
            }
        });

        executorService.shutdown();
    }
    public static DataFrame getHotelInfo(HiveContext hiveContext){
        String sql = "select * from common.dict_hotel_ol";
        return  hiveContext.sql(sql);
    }
}