spark streaming初体验

灰太狼 2022-08-18 03:25 8阅读 0赞

## spark streaming是建立在spark上的大规模实时流计算框架。通过阅读官方文档，做了一个本地的测试。 ##

**1.需要的工具**

netcat:通过TCP和UDP在网络中读写数据的工具
    下载地址：http://sourceforge.net/projects/netcat/files/netcat/0.7.1/netcat-0.7.1.tar.gz
    
    spark：编译安装并起动的spark，可以直接下载官方已经编译好的包，解压之后直接运行。（参考官方文档）
    
    主机：由于是在本地测试，只需要一台计算机。

**2.起动netcat的监听端口**

在“终端1”中输入：nc -l -p 9999
    [root@db1 netcat-0.7.1]# nc -l -p 9999
    起动成功后，光标在下一行一直闪烁等待输入数据。

**3.测试netcat**

另外打开一个终端，暂叫做“终端2”，在”终端2“中输入：nc -l -p 9999
    此时，”终端2“就等待”终端1“输入数据。我们在”终端1“中输入：12，后回车，在”终端2“中会看到“12”，表示收到数据。
    停止”终端2“的netcat，进行下一步的streaming测试。

**4.在”终端2“中运行spark作业**

如果没有配置环境变量，就切换目录到spark/bin目录下
    [root@db1 bin]# pwd
    /opt/spark/bin
    [root@db1 bin]# ./run-example streaming.NetworkWordCount localhost 9999
    （NetworkWordCount源代码在：spark/examples/src/main/scala/org/apache/spark/examples/streaming/NetworkWordCount.scala
    9999：是我们在”终端1“用netcat启动的监听端口）
    会车之后，会看到已下信息：
    .
    .
    .
    -------------------------------------------
    Time: 1447049226000 ms
    -------------------------------------------
    
    15/11/09 01:07:06 INFO scheduler.JobScheduler: Finished job streaming job 1447049226000 ms.0 from job set of time 1447049226000 ms
    15/11/09 01:07:06 INFO scheduler.JobScheduler: Total delay: 0.052 s for time 1447049226000 ms (execution: 0.047 s)
    15/11/09 01:07:06 INFO rdd.ShuffledRDD: Removing RDD 563 from persistence list
    15/11/09 01:07:06 INFO storage.BlockManager: Removing RDD 563
    15/11/09 01:07:06 INFO rdd.MapPartitionsRDD: Removing RDD 562 from persistence list
    15/11/09 01:07:06 INFO storage.BlockManager: Removing RDD 562
    15/11/09 01:07:06 INFO rdd.MapPartitionsRDD: Removing RDD 561 from persistence list
    15/11/09 01:07:06 INFO storage.BlockManager: Removing RDD 561
    15/11/09 01:07:06 INFO rdd.BlockRDD: Removing RDD 560 from persistence list
    15/11/09 01:07:06 INFO storage.BlockManager: Removing RDD 560
    15/11/09 01:07:06 INFO dstream.SocketInputDStream: Removing blocks of RDD BlockRDD[560] at socketTextStream at NetworkWordCount.scala:53 of time 1447049226000 ms
    15/11/09 01:07:06 INFO scheduler.ReceivedBlockTracker: Deleting batches ArrayBuffer(1447049224000 ms)
    .
    .

**5.在”终端1“中输入数据**

输入测试数据并以空格分割，比如输入：hi this is a test！
    在”终端2“中会看到如下输出：
    -------------------------------------------
    Time: 1447053798000 ms
    -------------------------------------------
    (this,1)
    (is,1)
    (a,1)
    (hi,1)
    (test！,1)