您好,登錄后才能下訂單哦!
Spark程序中一個Job的觸發(fā)是通過一個Action算子,比如count(), saveAsTextFile()等
在這次Spark優(yōu)化測試中,從Hive中讀取數(shù)據(jù),將其另外保存四份,其中兩個Job采用串行方式,另外兩個Job采用并行方式。將任務(wù)提交到Y(jié)arn中執(zhí)行。能夠明顯看出串行與兵線處理的性能。
每個Job執(zhí)行時間:
JobID | 開始時間 | 結(jié)束時間 | 耗時 |
Job 0 | 16:59:45 | 17:00:34 | 49s |
Job 1 | 17:00:34 | 17:01:13 | 39s |
Job 2 | 17:01:15 | 17:01:55 | 40s |
Job 3 | 17:01:16 | 17:02:12 | 56s |
四個Job都是自執(zhí)行相同操作,Job0,Job1一組采用串行方式,Job2,Job3采用并行方式。
Job0,Job1串行方式耗時等于兩個Job耗時之和 49s+39s=88s
Job2,Job3并行方式耗時等于最先開始和最后結(jié)束時間只差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ù)時要把這里放開 final DataFrame df = getHotelInfo(hiveContext); //沒有多線程處理的情況,連續(xù)執(zhí)行兩個Action操作,生成兩個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<Void>() { @Override public Void call(){ df.rdd().saveAsTextFile(rb.getString("hdfspath") + "/file3",com.hadoop.compression.lzo.LzopCodec.class); return null; } }); executorService.submit(new Callable<Void>() { @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); } }
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報,并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。