2. 程式人生 > >spark學習記錄（十三、SparkStreaming）

spark學習記錄（十三、SparkStreaming）

阿新 • • 發佈：2019-01-13

一、SparkStreaming簡介

SparkStreaming是流式處理框架，是Spark API的擴充套件，支援可擴充套件、高吞吐量、容錯的實時資料流處理，實時資料的來源可以是：Kafka, Flume, Twitter, ZeroMQ或者TCP sockets，並且可以使用高階功能的複雜運算元來處理流資料。例如：map,reduce,join,window 。最終，處理後的資料可以存放在檔案系統，資料庫等，方便實時展現。

SparkStreaming與Storm的區別:

  1. Storm是純實時的流式處理框架，SparkStreaming是準實時的處理框架（微批處理）。因為微批處理，SparkStreaming的吞吐量比Storm要高。Storm是來一條資料就處理一次，SparkStreaming是處理某段時間內來的資料。
  2. Storm 的事務機制要比SparkStreaming的要完善。
  3. Storm支援動態資源排程。(spark1.2開始和之後也支援)
  4. SparkStreaming擅長複雜的業務處理，Storm不擅長複雜的業務處理，擅長簡單的彙總型計算。

 

  • receiver  task是7*24小時一直在執行，一直接收資料，將一段時間內接收來的資料儲存到batch中。假設batchInterval為5s,那麼會將接收來的資料每隔5秒封裝到一個batch中，batch沒有分散式計算特性，這一個batch的資料又被封裝到一個RDD中，RDD最終封裝到一個DStream中。

例如：假設batchInterval為5秒，每隔5秒通過SparkStreamin將得到一個DStream,在第6秒的時候計算這5秒的資料，假設執行任務的時間是3秒,那麼第6~9秒一邊在接收資料，一邊在計算任務，9~10秒只是在接收資料。然後在第11秒的時候重複上面的操作。

  • 如果job執行的時間大於batchInterval會有什麼樣的問題？

如果接受過來的資料設定的級別是僅記憶體，接收來的資料會越堆積越多，最後可能會導致OOM（如果設定StorageLevel包含disk, 則記憶體存放不下的資料會溢寫至disk, 加大延遲 ） 

二、java程式碼

新增依賴：

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_2.12</artifactId>
    <version>2.4.0</version>
    <scope>provided</scope>
</dependency>

在hadoop1上輸入命令：

nc -lk 9999
hello Sam
hello Tom
hello Jetty
public class JavaExample {
    public static void main(String[] args) throws InterruptedException {
        SparkConf conf = new SparkConf();
        conf.setMaster("local[2]");
        conf.setAppName("SparkStreamingTest");

        JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));

        JavaReceiverInputDStream<String> stream = jsc.socketTextStream("192.168.30.141", 9999);

        final JavaDStream<String> words = stream.flatMap(new FlatMapFunction<String, String>() {
            public Iterator<String> call(String s) throws Exception {
                return Arrays.asList(s.split(" ")).iterator();
            }
        });
        JavaPairDStream<String, Integer> pair = words.mapToPair(new PairFunction<String, String, Integer>() {
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<String, Integer>(s, 1);
            }
        });

        JavaPairDStream<String, Integer> reduceByKey = pair.reduceByKey(new Function2<Integer, Integer, Integer>() {
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });

        //列印輸出
        reduceByKey.print();
        /**
         * foreachRDD可以拿到DStream中的RDD，對拿到的RDD可以使用RDD的transformations運算元轉換，
         * 要對拿到的RDD使用action觸發執行，否則foreachRDD也不會執行
         * foreachRDD中call方法內，拿到RDD的運算元外，程式碼在Driver執行，
         * 可以使用這個運算元實現動態改變廣播變數，即廣播讀取一個檔案，只修改檔案內容，不停止程式碼
         */
        reduceByKey.foreachRDD(new VoidFunction<JavaPairRDD<String, Integer>>() {
            public void call(JavaPairRDD<String, Integer> rdd) throws Exception {
                //SparkStreaming廣播變數
                SparkContext context = rdd.context();
                JavaSparkContext javaSparkContext = new JavaSparkContext(context);
                Broadcast<String> broadcast = javaSparkContext.broadcast("hello");
                System.out.println(broadcast.getValue());

                JavaPairRDD<String, Integer> mapToPair = rdd.mapToPair(new PairFunction<Tuple2<String, Integer>, String, Integer>() {
                    public Tuple2<String, Integer> call(Tuple2<String, Integer> tuple2) throws Exception {
                        return new Tuple2<String, Integer>(tuple2._1 + "~", tuple2._2);
                    }
                });
                mapToPair.foreach(new VoidFunction<Tuple2<String, Integer>>() {
                    public void call(Tuple2<String, Integer> tuple2) throws Exception {
                        System.out.println(tuple2);
                    }
                });
            }
        });
        jsc.start();
        jsc.awaitTermination();
    }
}

三、運算元 

/**
 * updateStateByKey:
 * 返回一個新的“狀態”Dstream,通過給定的func來更新之前的每個狀態的key對應的value值，這也可以用於維護key的任意狀態資料。
 * 注意：作用在（K,V）格式的DStream上
 * <p>
 * updateStateByKey的主要功能:
 * 1、Spark Streaming中為每一個Key維護一份state狀態，state型別可以是任意型別的的， 可以是一個自定義的物件，那麼更新函式也可以是自定義的。
 * 2、通過更新函式對該key的狀態不斷更新，對於每個新的batch而言，Spark Streaming會在使用updateStateByKey的時候為已經存在的key進行
 * state的狀態更新
 * （對於每個新出現的key，會同樣的執行state的更新函式操作），
 * 如果要不斷的更新每個key的state，就一定涉及到了狀態的儲存和容錯，這個時候就需要開啟checkpoint機制和功能
 *
 * @author root
 */
public class Operate_updateStateByKey {
    public static void main(String[] args) throws InterruptedException {
        SparkConf conf = new SparkConf().setMaster("local").setAppName("Operate_count");
        JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));
        /**
         * checkpoint儲存地址
         * 如果batchInterval小於10，那麼10s會將記憶體中的資料寫入硬碟中
         * 如果batchInterval大於10，那麼以batchInterval為準
         *
         * 這樣做是為了防止頻繁的寫hdfs
         */
        jsc.checkpoint("checkpoint");
        //讀取檔案儲存地址
        JavaDStream<String> textFileStream = jsc.textFileStream("data");
        /**
         * 實現一個累加統計word的功能
         */
        JavaPairDStream<String, Integer> mapToPair = textFileStream.flatMap(new FlatMapFunction<String, String>() {
            private static final long serialVersionUID = 1L;

            public Iterator<String> call(String t) throws Exception {

                return Arrays.asList(t.split(" ")).iterator();
            }
        }).mapToPair(new PairFunction<String, String, Integer>() {

            private static final long serialVersionUID = 1L;

            public Tuple2<String, Integer> call(String t) throws Exception {
                return new Tuple2<String, Integer>(t.trim(), 1);
            }
        });

        JavaPairDStream<String, Integer> updateStateByKey = mapToPair.updateStateByKey(new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() {
            private static final long serialVersionUID = 1L;

            public Optional<Integer> call(List<Integer> values, Optional<Integer> state)
                    throws Exception {
                /**
                 * values:經過分組最後 這個key所對應的value  [1,1,1,1,1]
                 * state:這個key在本次之前之前的狀態
                 */
                Integer updateValue = 0;

                if (state.isPresent()) {
                    updateValue = state.get();
                }
                for (Integer i : values) {
                    updateValue += i;
                }
                return Optional.of(updateValue);
            }
        });

        updateStateByKey.print();

        jsc.start();
        jsc.awaitTermination();
        jsc.close();
    }

}
public class Operate_window {
    public static void main(String[] args) throws InterruptedException {
        SparkConf conf = new SparkConf();
        conf.setMaster("local[2]");
        conf.setAppName("Operate_window");
        JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));
        /**
         * 設定日誌級別
         */
        jsc.sparkContext().setLogLevel("WARN");
        //設定checkpoint
        jsc.checkpoint("./checkpoint");
        JavaReceiverInputDStream<String> textStream = jsc.socketTextStream("192.168.30.141", 9999);
//        切分
        JavaDStream<String> flatMap = textStream.flatMap(new FlatMapFunction<String, String>() {
            public Iterator<String> call(String s) throws Exception {
                return Arrays.asList(s.split(" ")).iterator();
            }
        });
//        計數
        JavaPairDStream<String, Integer> mapToPair = flatMap.mapToPair(new PairFunction<String, String, Integer>() {
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<String, Integer>(s, 1);
            }
        });
        /**
         *每個10s，計算最近60s內的資料，那麼這個視窗大小為60s，裡面有12個rdd，在沒有計算之前，這些rdd是不會計算的
         * 那麼在計算的時候會將這12個rdd聚合起來，然後一起執行reduceByKeyAndWindow，
         * reduceByKeyAndWindow是針對視窗操作而不是DStream
         */
        JavaPairDStream<String, Integer> reduceByKeyAndWindow = mapToPair.reduceByKeyAndWindow(new Function2<Integer, Integer, Integer>() {
            public Integer call(Integer v1, Integer v2) throws Exception {
                //加上新進視窗的批次
                return v1 + v2;
            }
            //每隔5s計算最近15s的資料，這兩個引數要是bathInterval的整數倍即上面設定的5s的整數倍
        }, new Function2<Integer, Integer, Integer>() {
            public Integer call(Integer v1, Integer v2) throws Exception {
                //減去新離開視窗的批次
                return v1 - v2;
            }
        }, Durations.seconds(15), Durations.seconds(5));

        reduceByKeyAndWindow.print();
        jsc.start();
        jsc.awaitTermination();
        jsc.close();
    }
}

public class Operate_transform {
    public static void main(String[] args) throws InterruptedException {
        SparkConf conf = new SparkConf();
        conf.setMaster("local[2]").setAppName("Operate_transform");
        JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));
        //黑名單
        List<String> list = Arrays.asList("Sam");
        final Broadcast<List<String>> broadcast = jsc.sparkContext().broadcast(list);

        JavaReceiverInputDStream<String> stream = jsc.socketTextStream("192.168.30.141", 9999);

        JavaPairDStream<String, String> mapToPair = stream.mapToPair(new PairFunction<String, String, String>() {
            public Tuple2<String, String> call(String s) throws Exception {
                return new Tuple2<String, String>(s.split(" ")[1], s);
            }
        });
        /**
         * 過濾黑名單上的名字
         * transform可以拿到DStream中的RDD，做RDD到RDD之間的轉換，不需要Action運算元觸發，需要返回型別RDD
         * 注意：transform call方法內，拿到RDD運算元外的程式碼在Driver端執行，也可以做到動態改變廣播變數
         */
        JavaDStream<String> transform = mapToPair.transform(new Function<JavaPairRDD<String, String>, JavaRDD<String>>() {
            public JavaRDD<String> call(JavaPairRDD<String, String> nameRDD) throws Exception {
                //資料過濾
                JavaPairRDD<String, String> filter = nameRDD.filter(new Function<Tuple2<String, String>, Boolean>() {
                    public Boolean call(Tuple2<String, String> tuple2) throws Exception {
                        return !broadcast.getValue().contains(tuple2._1);
                    }
                });

                JavaRDD<String> map = filter.map(new Function<Tuple2<String, String>, String>() {
                    public String call(Tuple2<String, String> tuple2) throws Exception {
                        return tuple2._2;
                    }
                });

                return map;
            }
        });

        transform.print();

        jsc.start();
        jsc.awaitTermination();
        jsc.close();

    }
}

 

相關推薦

spark學習記錄（十三、SparkStreaming）

一、SparkStreaming簡介 SparkStreaming是流式處理框架，是Spark API的擴充套件，支援可擴充套件、高吞吐量、容錯的實時資料流處理，實時資料的來源可以是：Kafka, Flume, Twitter, ZeroMQ或者TCP sockets，並且可以使用高階功能的複雜

spark學習記錄（二、RDD）

一、概念 RDD（Resilient Distributed Dataset）叫做彈性分散式資料集，是Spark中最基本的資料抽象，它代表一個不可變、可分割槽、裡面的元素可平行計算的集合。RDD具有資料流模型的特點：自動容錯、位置感知性排程和可伸縮性。RDD允許使用者在執行多個查詢時顯式地將工作

spark學習記錄（十、SparkSQL）

一、介紹 SparkSQL支援查詢原生的RDD。 RDD是Spark平臺的核心概念，是Spark能夠高效的處理大資料的各種場景的基礎。 能夠在Scala中寫SQL語句。支援簡單的SQL語法檢查，能夠在Scala中寫Hive語句訪問Hive資料，並將結果取回作為RDD使用。 D

spark學習記錄（八、廣播變數和累加器）

一、廣播變數 public class JavaExample { public static void main(String[] args) { SparkConf conf = new SparkConf(); conf.setMaster("

spark學習記錄（九、MasterHA和Spark shuffle）

一、zookeeper配置MasterHA 1.1修改conf下的spark-env.sh ： export SPARK_DAEMON_JAVA_OPTS="-Dspark-deploy-recoveryMode=ZOOKEEPER -Dspark.deploy.zookee

spark學習記錄（七、二次排序和分組取TopN問題）

1.二次排序 例題：將兩列數字按第一列升序，如果第一列相同，則第二列升序排列 資料檔案：https://download.csdn.net/download/qq_33283652/10894807 將資料封裝成物件，對物件進行排序，然後取出value public class Se

spark學習記錄（六、基礎知識）

1.術語解釋 2.SparkCore和SparkSQL知識點思維導圖整理 https://download.csdn.net/download/qq_33283652/10890863  3.RDD的寬窄依賴 相同的key去同一個分割槽，但一個分割槽可以用不同的key

spark學習記錄（五、Spark基於資源排程管理器的提交模式）

一、Standalone（Spark自帶） 1.1 Standalone-client模式  提交命令： ./spark-submit --master spark://hadoop1:7077 --class org.apache.spark.examples.Spar

spark學習記錄（四、運算元（函式））

1.Transformations轉換運算元 Transformations類運算元是一類運算元（函式）叫做轉換運算元，如map,flatMap,reduceByKey等。Transformations運算元是延遲執行，也叫懶載入執行。 filter：過濾符合條件的記錄數，true保留

spark學習記錄（三、spark叢集搭建）

一、安裝spark 1.上傳壓縮包並解壓 2.在conf目錄下配置slaves cp slaves.template slaves //在master機上配置worker節點 hadoop2 hadoop3 3.配置spark-env.sh cp spark-env.sh.t

spark學習記錄（一、scala與java編寫wordCount比較）

 新增依賴： <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.12<

Hadoop學習記錄（三、MapReduce）

1.將一個日誌檔案上傳到hdfs上   2. 編寫mapReduce程式碼 2.1新建一個maven專案，新增依賴 <dependencies> <dependency> <groupId>

Storm學習記錄（一、簡介）

一、簡介 Storm是一個免費並開源的分散式實時計算系統。利用Storm可以很容易做到可靠地處理無限的資料流，像Hadoop批量處理大資料一樣，Storm可以實時處理資料。Storm簡單，可以使用任何程式語言。 Storm有如下特點： 程式設計簡單：開發人員只需要關注應用邏輯，而且

spark學習記錄（十四、kafka）

一、簡介 kafka是一個高吞吐的分散式訊息佇列系統。特點是生產者消費者模式，先進先出（FIFO）保證順序，自己不丟資料，預設每隔7天清理資料。訊息列隊常見場景：系統之間解耦合、峰值壓力緩衝、非同步通訊。 Kafka架構是由producer（訊息生產者）、consumer（訊息消費者）

spark學習記錄（十二、Spark UDF&UDAF&開窗函式）

一、UDF&UDAF  public class JavaExample { public static void main(String[] args) { SparkConf conf = new SparkConf(); conf.

spark學習記錄（十一、Spark on Hive配置）

新增依賴 <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-hive_2.12</artifa

Hadoop學習記錄（五、hadoop IO操作）

1.壓縮從標準輸入讀取的資料，然後將其寫到標準輸出 通過GzipCodec的StreamCompressor物件對字串“Text”進行壓縮，再使用gunzip從標準輸出中對它進行讀取並解壓縮 public class StreamCompressor { public static

Hadoop學習記錄（四、hadoop實現檔案操作）

1.從Hadoop URL讀取資料 類似cat命令 public class URLCat { static{ URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory()); }

Hadoop學習記錄（二、hdfs shell命令）

在/usr/local/hadoop-2.9.2/bin目錄下執行命令 1.檢視根目錄： ./hdfs dfs -ls / 2.檔案上傳：上傳到根目錄 ./hdfs dfs -put /tmp/test.txt /   3.檢視檔案內容 ./hdfs df

Hadoop學習記錄（一、Hadoop叢集的搭建）

參考：http://www.zuidemo.com/filePreview/pdfFilePreview/11202並進行補充 1.新建七個centos7系統的虛擬機器，分別命名為cluster1,cluster2等。關閉防火牆。 2.七臺主機都修改host檔案 vi /etc/host