欧美一区二区三区老妇人-欧美做爰猛烈大尺度电-99久久夜色精品国产亚洲a-亚洲福利视频一区二区

SparkStreaming是什么

這篇文章主要講解了“Spark Streaming是什么”，文中的講解內容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“Spark Streaming是什么”吧！

十多年的大通網站建設經驗，針對設計、前端、開發(fā)、售后、文案、推廣等六對一服務，響應快，48小時及時工作處理。營銷型網站建設的優(yōu)勢是能夠根據用戶設備顯示端的尺寸不同，自動調整大通建站的顯示方式，使網站能夠適用不同顯示終端，在瀏覽器中調整網站的寬度，無論在任何一種瀏覽器上瀏覽網站，都能展現(xiàn)優(yōu)雅布局與設計，從而大程度地提升瀏覽體驗。成都創(chuàng)新互聯(lián)從事“大通網站設計”,“大通網站推廣”以來，每個客戶項目都認真落實執(zhí)行。

一：Spark Streaming 概覽。

1.1    簡單了解 Spark Streaming。

         Spark Streaming 是核心 Spark API的一個擴展。具有可擴展性，高吞吐量，容錯性，實時性等特征。

        數(shù)據從許多來如中攝入數(shù)據，如 Kafka, Flume, Twitter, ZeroMQ, Kinesis, or TCP sockets。

也可以使用復雜的算法與高級別的功能像map，reduce，join和window處理。

        最后，也可以將處理過的數(shù)據推送到文件系統(tǒng)、數(shù)據庫。事實上，我們也可以用Spark 的機器學習和圖形處理數(shù)據流上的算法。用圖形表示如下：Spark Streaming是什么
        在內部，其工作原理如下。Spark Streaming接收實時輸入的數(shù)據流和數(shù)據劃分批次，然后由Spark引擎批處理生成的最終結果流。如圖示：  Spark Streaming是什么

      另外，Spark Streaming提供一個高級抽象，稱為離散的流或 DStream，表示連續(xù)的流的數(shù)據。DStreams 可以被創(chuàng)建從輸入的數(shù)據流，如Kafka, Flume, and Kinesis,

        或采用其他的DStreams高級別操作的輸入的數(shù)據流。

        在內部，DStream 是以 RDDs 的序列來表示。

首先，看看Maven的依賴包（spark-streaming_2.10）管理：

        <dependency>
			<groupId>org.apache.spark</groupId>
			<artifactId>spark-streaming_2.10</artifactId>
			<version>1.6.1</version>
		</dependency>

1.2    eg：從一個數(shù)據服務器監(jiān)聽 TCP 套接字接收的文本數(shù)據中的單詞進行計數(shù)

package com.berg.spark.test5.streaming;

import java.util.Arrays;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

public class SparkStreamingDemo1 {

	public static void main(String[] args) {

		SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount");
		conf.set("spark.testing.memory", "269522560000");

		JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(10));
		System.out.println("jssc: " + jssc);

		// 創(chuàng)建一個DStream, 將連接 hostname:port, 比如 master:9999
		JavaReceiverInputDStream<String> lines = jssc.socketTextStream("master", 9999);
		System.out.println("lines : " + lines);

		JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {

			private static final long serialVersionUID = 1L;

			public Iterable<String> call(String x) {
				return Arrays.asList(x.split(" "));
			}
		});

		// Count each word in each batch
		JavaPairDStream<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
			public Tuple2<String, Integer> call(String s) {
				return new Tuple2<String, Integer>(s, 1);
			}
		});
		JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
			public Integer call(Integer i1, Integer i2) {
				return i1 + i2;
			}
		});

		// Print the first ten elements of each RDD generated in this DStream to
		// the console
		wordCounts.print();

		jssc.start();  // Start the computation
		jssc.awaitTermination();  // Wait for the computation to terminate
	}
}

至于代碼如何運行了，首先在Linux下終端輸入：$ nc -lk 9999

然后在Eclipse中運行代碼 。

隨意輸入一行文本單詞，單詞之間用空格隔開，如下：

hadoop@master:~$ nc -lk 9999
berg hello world berg hello

可以在Eclipse控制臺看到如下結果：

Time: 1465386060000 ms
-------------------------------------------
(hello,2)
(berg,2)
(world,1)

1.3 將HDFS目錄下的某些文件內容當做 輸入的數(shù)據流。

public class SparkStreamingDemo2 {

	public static void main(String[] args) {

		SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount");
		conf.set("spark.testing.memory", "269522560000");

		JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(10));
		System.out.println("jssc: " + jssc);

		// 創(chuàng)建一個DStream, 讀取HDFS上的文件,作為數(shù)據源。
		JavaDStream<String> lines = jssc.textFileStream("hdfs://master:9000/txt/sparkstreaming/");

		System.out.println("lines : " + lines);

		// Split each line into words
		JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
			private static final long serialVersionUID = 1L;

			public Iterable<String> call(String x) {
				System.out.println(Arrays.asList(x.split(" ")).get(0));
				return Arrays.asList(x.split(" "));
			}
		});

		// Count each word in each batch
		JavaPairDStream<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
			private static final long serialVersionUID = 1L;

			public Tuple2<String, Integer> call(String s) {
				return new Tuple2<String, Integer>(s, 1);
			}
		});
		System.out.println(pairs);
		
		JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
			public Integer call(Integer i1, Integer i2) {
				return i1 + i2;
			}
		});

		// Print the first ten elements of each RDD generated in this DStream to the console
		wordCounts.print();
		
		JavaDStream<Long> count = wordCounts.count();
		count.print(); // 統(tǒng)計
		
		DStream<Tuple2<String, Integer>>  dstream = wordCounts.dstream();
		dstream.saveAsTextFiles("hdfs://master:9000/bigdata/spark/xxxx", "sparkstreaming");
		//wordCounts.dstream().saveAsTextFiles("hdfs://master:9000/bigdata/spark/xxxx", "sparkstreaming");
		
		jssc.start();  
		jssc.awaitTermination();   // Wait for the computation to terminate
	}
}

上述代碼完成的操作是，一直監(jiān)聽HDFS即hdfs://master:9000/txt/sparkstreaming/目錄下是否有文件存入，如果有，則統(tǒng)計文件中的單詞。。。。

嘗試運行程序后，然后往該目錄中手動添加一個文件，會在控制臺看到對該文件內容中的單詞統(tǒng)計后的數(shù)據。

注意參數(shù)的意義：


 public JavaDStream<java.lang.String> textFileStream(java.lang.String directory)
  Create an input stream that monitors a Hadoop-compatible filesystem for 
             new files and reads them as text 
                      files (using key as LongWritable, value as Text and input format as TextInputFormat).
                 Files must be written to the monitored directory 
                 by "moving" them from another location within the same file system. 
                 File names starting with . are ignored.
 Parameters:
 directory - HDFS directory to monitor for new file
 Returns:
 (undocumented)

感謝各位的閱讀，以上就是“Spark Streaming是什么”的內容了，經過本文的學習后，相信大家對Spark Streaming是什么這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是創(chuàng)新互聯(lián)，小編將為大家推送更多相關知識點的文章，歡迎關注！

當前題目：SparkStreaming是什么
當前地址：http://chinadenli.net/article30/giijso.html

成都網站建設公司_創(chuàng)新互聯(lián)，為您提供商城網站、外貿建站、網站改版、微信小程序、網站收錄、網站設計

廣告

聲明：本網站發(fā)布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經允許不得轉載，或轉載時需注明來源： 創(chuàng)新互聯(lián)

成都seo排名網站優(yōu)化