ITPub博客

首页 > 大数据 > Hadoop > Spark Stream 简单例子

Spark Stream 简单例子

原创 Hadoop 作者:21ca 时间:2017-03-31 16:03:38 0 删除 编辑
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. import org.apache.spark.SparkConf;
  7. import org.apache.spark.streaming.Durations;
  8. import org.apache.spark.streaming.api.java.JavaDStream;
  9. import org.apache.spark.streaming.api.java.JavaPairDStream;
  10. import org.apache.spark.streaming.api.java.JavaStreamingContext;

  11. import scala.Tuple2;

  12. public class SparkStreamTest {

  13.     public static void main(String[] args) throws Exception {
  14.         startSockerServer(9999);
  15.         
  16.         SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount");
  17.         JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));
  18.         JavaDStream<String> lines = jssc.socketTextStream("localhost", 9999);

  19.         JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(x.split(" ")).iterator());
  20.         JavaPairDStream<String, Integer> pairs = words.mapToPair(s -> new Tuple2<>(s, 1));
  21.         JavaPairDStream<String, Integer> counts = pairs.reduceByKey((x, y) -> x + y);
  22.         counts.print();
  23.         
  24.         jssc.start();
  25.         jssc.awaitTermination();
  26.         jssc.close();
  27.     }

  28.     private static void startSockerServer(int port) throws Exception {
  29.         new Thread(){
  30.             public void run() {
  31.                 try {
  32.                     ServerSocket server = new ServerSocket(port);
  33.                     Socket socket = server.accept();
  34.                     OutputStream os = socket.getOutputStream();
  35.                     for (int i = 0; i < 1000; i++) {
  36.                         String txt = UUID.randomUUID().toString().replaceAll("-", " ");
  37.                         txt = "Hello test\n";
  38.                         os.write(txt.getBytes());
  39.                         Thread.sleep(10);
  40.                     }
  41.                     socket.close();
  42.                     server.close();
  43.                 } catch (Exception e) {
  44.                     e.printStackTrace();
  45.                 }
  46.             }
  47.         }.start();
  48.     }
  49. }
会不断地打印出类似的结果:
-------------------------------------------
Time: 1490947248000 ms
-------------------------------------------
(Hello,99)
(test,99)



来自 “ ITPUB博客 ” ,链接:http://blog.itpub.net/10742815/viewspace-2136411/,如需转载,请注明出处,否则将追究法律责任。

下一篇: Appium笔记
请登录后发表评论 登录
全部评论

注册时间:2009-03-24

  • 博文量
    58
  • 访问量
    83421