大資料實時計算Spark學習筆記(11)—— Spark Streaming
阿新 • • 發佈:2018-12-31
1 Spark Streaming
- spark core 的擴充套件,針對實時資料處理,具有可擴充套件、高吞吐、容錯;
- 內部,spark 接受實時資料流,分成 batch 進行處理,最終在每個 batch 產生結果;
1.1 discretized stream or DStream
- 通過kafka,flume 等輸入產生,或者通過其他的 DStream 進行高階變換產生;
- 在內部,DStream 表現為 RDD 序列;
2 Spark Streaming 測試案例
- POM 新增依賴
<dependency >
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.11</artifactId>
<version>${spark.version}</version>
</dependency>
2.1 Scala 流式單詞統計
package sparkstreaming
import org.apache.spark.SparkConf
import org.apache.spark. streaming.{Seconds, StreamingContext}
object StramingWordCount {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local[4]").setAppName("NetWordCount")
val ssc = new StreamingContext(conf, Seconds(10))
val lines = ssc.socketTextStream("localhost" , 9999)
val words = lines.flatMap(_.split(" "))
val pairs = words.map((_, 1))
val count = pairs.reduceByKey(_ + _)
count.print
ssc.start()
ssc.awaitTermination()
}
}
2.2 Java 版流式單詞統計
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;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
public class JavaStreamingWordcount {
public static void main(String[] args) throws InterruptedException {
SparkConf conf = new SparkConf().setAppName("JavaStreamingWordcount").setMaster("local[2]");
JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));
JavaReceiverInputDStream sock = jsc.socketTextStream("localhost", 9999);
JavaDStream<String> wordsDS = sock.flatMap(new FlatMapFunction<String, String>() {
@Override
public Iterator call(String str) throws Exception {
List<String> list = new ArrayList<String>();
String[] arr = str.split(" ");
for (String s : arr) {
list.add(s);
}
return list.iterator();
}
});
JavaPairDStream<String, Integer> pairDS = wordsDS.mapToPair(new PairFunction<String, String, Integer>() {
@Override
public Tuple2<String, Integer> call(String s) throws Exception {
return new Tuple2<String, Integer>(s, 1);
}
});
JavaPairDStream<String, Integer> countDS = pairDS.reduceByKey(new Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1 + v2;
}
});
countDS.print();
jsc.start();
jsc.awaitTermination();
}
}