Spark Stream怎么使用

这篇文章主要讲解了"Spark Stream怎么使用"，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习"Spark Stream怎么使用"吧！

pom.xml

1. org.apache.spark
   

2. spark-streaming_2.11
   

3. 2.1.0
   

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 lines = jssc.socketTextStream("localhost", 9999);
    

23. 
    

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

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

26. JavaPairDStream 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怎么使用这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是，小编将为大家推送更多相关知识点的文章，欢迎关注！