Spark Stream怎么使用

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

pom.xml
<dependency>

1.     <groupId>org.apache.spark</groupId>
   

2.     <artifactId>spark-streaming_2.11</artifactId>
   

3.     <version>2.1.0</version>
   

4. </dependency>

1. import java.io.OutputStream;
   

2. import java.net.ServerSocket;
   

3. import java.net.Socket;
   

4. import java.util.Arrays;
   

5. import java.util.UUID;
   

6. 
   

7. import org.apache.spark.SparkConf;
   

8. import org.apache.spark.streaming.Durations;
   

9. import org.apache.spark.streaming.api.java.JavaDStream;
   

10. import org.apache.spark.streaming.api.java.JavaPairDStream;
    

11. import org.apache.spark.streaming.api.java.JavaStreamingContext;
    

12. 
    

13. import scala.Tuple2;
    

14. 
    

15. public class SparkStreamTest {
    

16. 
    

17.     public static void main(String[] args) throws Exception {
    

18.         startSockerServer(9999);
    

19.         
    

20.         SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount");
    

21.         JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));
    

22.         JavaDStream<String> lines = jssc.socketTextStream("localhost", 9999);
    

23. 
    

24.         JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(x.split(" ")).iterator());
    

25.         JavaPairDStream<String, Integer> pairs = words.mapToPair(s -> new Tuple2<>(s, 1));
    

26.         JavaPairDStream<String, Integer> counts = pairs.reduceByKey((x, y) -> x + y);
    

27.         counts.print();
    

28.         
    

29.         jssc.start();
    

30.         jssc.awaitTermination();
    

31.         jssc.close();
    

32.     }
    

33. 
    

34.     private static void startSockerServer(int port) throws Exception {
    

35.         new Thread(){
    

36.             public void run() {
    

37.                 try {
    

38.                     ServerSocket server = new ServerSocket(port);
    

39.                     Socket socket = server.accept();
    

40.                     OutputStream os = socket.getOutputStream();
    

41.                     for (int i = 0; i < 1000; i++) {
    

42.                         String txt = UUID.randomUUID().toString().replaceAll("-", " ");
    

43.                         txt = "Hello test\n";
    

44.                         os.write(txt.getBytes());
    

45.                         Thread.sleep(10);
    

46.                     }
    

47.                     socket.close();
    

48.                     server.close();
    

49.                 } catch (Exception e) {
    

50.                     e.printStackTrace();
    

51.                 }
    

52.             }
    

53.         }.start();

54.     }

55. }

會不斷地打印出類似的結果：
-------------------------------------------
Time: 1490947248000 ms
-------------------------------------------
(Hello,99)
(test,99)

感謝各位的閱讀，以上就是“Spark Stream怎么使用”的內容了，經過本文的學習后，相信大家對Spark Stream怎么使用這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！