// 创建StreamingContext，1秒一个批次
val ssc = new StreamingContext(sparkConf, Seconds(1));

// 获得一个DStream负责连接 监听端口:地址
val lines = ssc.socketTextStream(serverIP, serverPort);

// 对每一行数据执行Split操作
val words = lines.flatMap(_.split(" "));
// 统计word的数量
val pairs = words.map(word => (word, 1));
val wordCounts = pairs.reduceByKey(_ + _);

// 输出结果
wordCounts.print();

ssc.start(); // 开始
ssc.awaitTermination(); // 计算完毕退出

$ nc -lk 9999

$ ./bin/run-example org.apache.spark.streaming.examples.NetworkWordCount local[2] localhost 9999

# TERMINAL 1:
# Running Netcat

$ nc -lk 9999

hello world

...
# TERMINAL 2: RUNNING NetworkWordCount or JavaNetworkWordCount

$ ./bin/run-example org.apache.spark.streaming.examples.NetworkWordCount local[2] localhost 9999
...
-------------------------------------------
Time: 1357008430000 ms
-------------------------------------------
(hello,1)
(world,1)
...

groupId = org.apache.spark
artifactId = spark-streaming_2.10
version = 0.9.0-incubating

//需要使用一下数据源的，还要添加相应的依赖
Source    Artifact
Kafka     spark-streaming-kafka_2.10
Flume     spark-streaming-flume_2.10
Twitter     spark-streaming-twitter_2.10
ZeroMQ     spark-streaming-zeromq_2.10
MQTT     spark-streaming-mqtt_2.10

new StreamingContext(master, appName, batchDuration, [sparkHome], [jars])

streamingContext.fileStream(dataDirectory)

import org.apache.spark.streaming.kafka._
KafkaUtils.createStream(streamingContext, kafkaParams, ...)

Transformation                          Meaning
map(func)                        对每一个元素执行func方法
flatMap(func)                    类似map函数，但是可以map到0+个输出
filter(func)                     过滤
repartition(numPartitions)       增加分区，提高并行度     
union(otherStream)               合并两个流
count()     　　　　　　　　       统计元素的个数
reduce(func)                     对RDDs里面的元素进行聚合操作，2个输入参数，1个输出参数
countByValue()                   针对类型统计，当一个Dstream的元素的类型是K的时候，
调用它会返回一个新的Dstream，包含<K,Long>键值对，Long是每个K出现的频率。
reduceByKey(func, [numTasks])    对于一个(K, V)类型的Dstream，为每个key，
执行func函数，默认是local是2个线程，cluster是8个线程，也可以指定numTasks 
join(otherStream, [numTasks])    把(K, V)和(K, W)的Dstream连接成一个(K, (V, W))的新Dstream 
cogroup(otherStream, [numTasks]) 把(K, V)和(K, W)的Dstream连接成一个(K, Seq[V], Seq[W])的新Dstream 
transform(func)                  转换操作，把原来的RDD通过func转换成一个新的RDD
updateStateByKey(func)           针对key使用func来更新状态和值，可以将state该为任何值

def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
    val newCount = ...  // add the new values with the previous running count to get the new count
    Some(newCount)
}

val runningCounts = pairs.updateStateByKey[Int](updateFunction _)

val spamInfoRDD = sparkContext.hadoopFile(...) // RDD containing spam information

val cleanedDStream = inputDStream.transform(rdd => {
rdd.join(spamInfoRDD).filter(...) // join data stream with spam information to do data cleaning
...
})

// Reduce last 30 seconds of data, every 10 seconds
val windowedWordCounts = pairs.reduceByKeyAndWindow(_ + _, Seconds(30), Seconds(10))

Transformation Meaning
window(windowLength, slideInterval)     
countByWindow(windowLength, slideInterval)     
reduceByWindow(func, windowLength, slideInterval)     
reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])     
reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks])    
countByValueAndWindow(windowLength, slideInterval, [numTasks])

Output Operation                                      Meaning
print()                                 打印到控制台
foreachRDD(func)                        对Dstream里面的每个RDD执行func，保存到外部系统
saveAsObjectFiles(prefix, [suffix])     保存流的内容为SequenceFile, 文件名 : "prefix-TIME_IN_MS[.suffix]".
saveAsTextFiles(prefix, [suffix])       保存流的内容为文本文件, 文件名 : "prefix-TIME_IN_MS[.suffix]".
saveAsHadoopFiles(prefix, [suffix])     保存流的内容为hadoop文件, 文件名 : "prefix-TIME_IN_MS[.suffix]".

ssc.checkpoint(hdfsPath)  //设置检查点的保存位置
dstream.checkpoint(checkpointInterval)  //设置检查点间隔

// Function to create and setup a new StreamingContext
def functionToCreateContext(): StreamingContext = {
    val ssc = new StreamingContext(...)   // new context
    val lines = ssc.socketTextStream(...) // create DStreams
    ...
    ssc.checkpoint(checkpointDirectory)   // set checkpoint directory
    ssc
}

// Get StreaminContext from checkpoint data or create a new one
val context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext _)

// Do additional setup on context that needs to be done,
// irrespective of whether it is being started or restarted
context. ...

// Start the context
context.start()
context.awaitTermination()

./bin/spark-class org.apache.spark.deploy.Client launch
   [client-options] \
   <cluster-url> <application-jar-url> <main-class> \
   [application-options]

cluster-url: master的地址.
application-jar-url: jar包的地址，最好是hdfs上的,带上hdfs：//...否则要所有的节点的目录下都有这个jar的
main-class: 要发布的程序的main函数所在类.
Client Options:
--memory <count> (驱动程序的内存，单位是MB)
--cores <count> (为你的驱动程序分配多少个核心)
--supervise (节点失效的时候，是否重新启动应用)
--verbose (打印增量的日志输出)

Time Number of lines in input file Output without driver failure Output with driver failure
　　10 　　　　10 　　　　　　10
　　20 　　　　20 　　　　　    20
　　30 　　　　30 　　　　　    30
　　40 　　　　40 　　　　　　[DRIVER FAILS] no output
　　50 　　　　50 　　　　　　no output
　　60 　　　　60 　　　　　　no output
　　70 　　　　70 　　　　　　[DRIVER RECOVERS] 40, 50, 60, 70
　　80 　　　　80 　　　　　　80
　　90 　　　　90 　　　　　　90
  　100 　　　　100 　　　　　  100