第一个Spark Streaming例子

        Spark Streaming是一种近实时的流式计算模型，它将作业分解成一批一批的短小的批处理任务，然后并行计算，具有可扩展，高容错，高吞吐，实时性高等一系列优点，在某些场景可达到与Storm一样的处理程度或优于storm，也可以无缝集成多重日志收集工具或队列中转器，比如常见的 kakfa，flume，redis，logstash等，计算完后的数据结果，也可以 存储到各种存储系统中，如HDFS，数据库等，一张简单的数据流图如下：

Spark Streaming内部处理流程:Spark Streaming内部处理流程:

接下来我们以NetworkWordCount为例，掌握Spark Streaming该如何使用。

package org.apache.spark.examples.streaming

import org.apache.spark.SparkConf

import org.apache.spark.storage.StorageLevel

import org.apache.spark.streaming.{Seconds, StreamingContext}

object NetworkWordCount {

  def main(args: Array[String]) {

    if (args.length < 2) {

      System.err.println("Usage: NetworkWordCount  ")

      System.exit(1)

    }

    StreamingExamples.setStreamingLogLevels()

    // Create the context with a 1 second batch size

    val sparkConf = new SparkConf().setAppName("NetworkWordCount")

    val ssc = new StreamingContext(sparkConf, Seconds(1))

    // Create a socket stream on target ip:port and count the

    // words in input stream of \n delimited text (eg. generated by 'nc')

    // Note that no duplication in storage level only for running locally.

    // Replication necessary in distributed scenario for fault tolerance.

    val lines = ssc.socketTextStream(args(0), args(1).toInt, StorageLevel.MEMORY_AND_DISK_SER)

    val words = lines.flatMap(_.split(" "))

    val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)

    wordCounts.print()

    ssc.start()

    ssc.awaitTermination()

  }

}

在spark 集群上启动NetworkWordCount。

然后在对应的linux机器上，开一个nc服务，并写入一些数据：

然后在控制台，可见计算结果，并且是排好序的：

至此，第一个Spark Streaming的demo就入门了，大家可以自己动手进行扩展，比如从kakfa或者redis里面接受数据，然后存储到hbase，或者mysql或者solr，lucene，elasticsearch索引中。