<译>Spark Sreaming 编程指南

importorg.apache.spark.*;
importorg.apache.spark.api.java.function.*;
importorg.apache.spark.streaming.*;
importorg.apache.spark.streaming.api.java.*;
importscala.Tuple2;

//CreatealocalStreamingContextwithtwoworkingthreadandbatchintervalof1second
SparkConfconf=newSparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
JavaStreamingContextjssc=newJavaStreamingContext(conf,Durations.seconds(1))

//CreateaDStreamthatwillconnecttohostname:port,likelocalhost:9999
JavaReceiverInputDStream<String>lines=jssc.socketTextStream("localhost",9999);

lines

//Spliteachlineintowords
JavaDStream<String>words=lines.flatMap(
newFlatMapFunction<String,String>(){
@OverridepublicIterable<String>call(Stringx){
returnArrays.asList(x.split(""));
}
});

//Counteachwordineachbatch
JavaPairDStream<String,Integer>pairs=words.mapToPair(
newPairFunction<String,String,Integer>(){
@OverridepublicTuple2<String,Integer>call(Strings){
returnnewTuple2<String,Integer>(s,1);
}
});
JavaPairDStream<String,Integer>wordCounts=pairs.reduceByKey(
newFunction2<Integer,Integer,Integer>(){
@OverridepublicIntegercall(Integeri1,Integeri2){
returni1+i2;
}
});

//PrintthefirsttenelementsofeachRDDgeneratedinthisDStreamtotheconsole
wordCounts.print();

words

(word,1)

wordCounts.print()

ssc.start()//Startthecomputation
ssc.awaitTermination()//Waitforthecomputationtoterminate

$nc-lk9999

$./bin/run-examplestreaming.NetworkWordCountlocalhost9999

#TERMINAL1:
#RunningNetcat

$nc-lk9999

helloworld

...

#TERMINAL2:RUNNINGNetworkWordCount

$./bin/run-examplestreaming.NetworkWordCountlocalhost9999
...
-------------------------------------------
Time:1357008430000ms
-------------------------------------------
(hello,1)
(world,1)
...

<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.10</artifactId>
<version>1.6.0</version>
</dependency>

spark-streaming-xyz_2.10

importorg.apache.spark._
importorg.apache.spark.streaming._

valconf=newSparkConf().setAppName(appName).setMaster(master)
valssc=newStreamingContext(conf,Seconds(1))

appName

master

master

spark-submit

ssc.sparkContext

StreamingContext

SparkContext

importorg.apache.spark.streaming._

valsc=...//existingSparkContext
valssc=newStreamingContext(sc,Seconds(1))

streamingContext.start()

streamingContext.awaitTermination()

streamingContext.stop()

stop()

stopSparkContext

streamingContext.fileStream<KeyClass,ValueClass,InputFormatClass>(dataDirectory);

actorStream

importorg.apache.spark.streaming.twitter.*;

TwitterUtils.createStream(jssc);

updateStateByKey

None

defupdateFunction(newValues:Seq[Int],runningCount:Option[Int]):Option[Int]={
valnewCount=...//addthenewvalueswiththepreviousrunningcounttogetthenewcount
Some(newCount)
}

pairs

(word,1)

valrunningCounts=pairs.updateStateByKey[Int](updateFunction_)

newValues

(word,1)

runningCount

updateStateByKey

transform

transformWith

transform

valspamInfoRDD=ssc.sparkContext.newAPIHadoopRDD(...)//RDDcontainingspaminformation

valcleanedDStream=wordCounts.transform(rdd=>{
rdd.join(spamInfoRDD).filter(...)//joindatastreamwithspaminformationtododatacleaning
...
})

reduceByKey

pairs

(word,1)

reduceByKeyAndWindow

//Reducelast30secondsofdata,every10seconds
valwindowedWordCounts=pairs.reduceByKeyAndWindow((a:Int,b:Int)=>(a+b),Seconds(30),Seconds(10))

spark.default.parallelism

numTasks

reduceByKeyAndWindow()

reduceByKeyAndWindow

reduceByKeyAndWindow

valstream1:DStream[String,String]=...
valstream2:DStream[String,String]=...
valjoinedStream=stream1.join(stream2)

stream1

stream2

leftOuterJoin

rightOuterJoin

fullOuterJoin

valwindowedStream1=stream1.window(Seconds(20))
valwindowedStream2=stream2.window(Minutes(1))
valjoinedStream=windowedStream1.join(windowedStream2)

DStream.transform

valdataset:RDD[String,String]=...
valwindowedStream=stream.window(Seconds(20))...
valjoinedStream=windowedStream.transform{rdd=>rdd.join(dataset)}

transform

dataset

SequenceFiles

dstream.foreachRDD

dstream.foreachRDD{rdd=>
valconnection=createNewConnection()//executedatthedriver
rdd.foreach{record=>
connection.send(record)//executedattheworker
}
}

dstream.foreachRDD{rdd=>
rdd.foreach{record=>
valconnection=createNewConnection()
connection.send(record)
connection.close()
}
}

rdd.foreachPartition

dstream.foreachRDD{rdd=>
rdd.foreachPartition{partitionOfRecords=>
valconnection=createNewConnection()
partitionOfRecords.foreach(record=>connection.send(record))
connection.close()
}
}

dstream.foreachRDD{rdd=>
rdd.foreachPartition{partitionOfRecords=>
//ConnectionPoolisastatic,lazilyinitializedpoolofconnections
valconnection=ConnectionPool.getConnection()
partitionOfRecords.foreach(record=>connection.send(record))
ConnectionPool.returnConnection(connection)//returntothepoolforfuturereuse
}
}

dstream.foreachRDD()

/**DataFrameoperationsinsideyourstreamingprogram*/

valwords:DStream[String]=...

words.foreachRDD{rdd=>

//GetthesingletoninstanceofSQLContext
valsqlContext=SQLContext.getOrCreate(rdd.sparkContext)
importsqlContext.implicits._

//ConvertRDD[String]toDataFrame
valwordsDataFrame=rdd.toDF("word")

//Registerastable
wordsDataFrame.registerTempTable("words")

//DowordcountonDataFrameusingSQLandprintit
valwordCountsDataFrame=
sqlContext.sql("selectword,count(*)astotalfromwordsgroupbyword")
wordCountsDataFrame.show()
}

streamingContext.remember(Minutes(5))

persist()

reduceByWindow

reduceByKeyAndWindow

updateStateByKey

persist()

updateStateByKey

reduceByKeyAndWindow

streamingContext.checkpoint(checkpointDirectory)

StreamingContext.getOrCreate

//FunctiontocreateandsetupanewStreamingContext
deffunctionToCreateContext():StreamingContext={
valssc=newStreamingContext(...)//newcontext
vallines=ssc.socketTextStream(...)//createDStreams
...
ssc.checkpoint(checkpointDirectory)//setcheckpointdirectory
ssc
}

//GetStreamingContextfromcheckpointdataorcreateanewone
valcontext=StreamingContext.getOrCreate(checkpointDirectory,functionToCreateContext_)

//Doadditionalsetuponcontextthatneedstobedone,
//irrespectiveofwhetheritisbeingstartedorrestarted
context....

//Startthecontext
context.start()
context.awaitTermination()

checkpointDirectory

functionToCreateContext

getOrCreate

dstream.checkpoint(checkpointInterval)

spark-submit

TwitterUtils

spark-streaming-twitter_2.10

spark.streaming.receiver.writeAheadLog.enable

true

StorageLevel.MEMORY_AND_DISK_SER

spark.streaming.receiver.maxRate

spark.streaming.kafka.maxRatePerPartition

spark.streaming.backpressure.enabled

true

StreamingContext.stop(...)

JavaStreamingContext.stop(...)

spark.streaming.receiver.maxRate

Streaming

valnumStreams=5
valkafkaStreams=(1tonumStreams).map{i=>KafkaUtils.createStream(...)}
valunifiedStream=streamingContext.union(kafkaStreams)
unifiedStream.print()

spark.streaming.blockInterval

inputStream.repartition(<numberofpartitions>)

reduceByKey

reduceByKeyAndWindow

spark.default.parallelism

PairDStreamFunctions

spark.default.parallelism

spark.closure.serializer

updateStateByKey

spark.rdd.compress

streamingContext.remember

--driver-java-options

spark-submit

spark.executor.extraJavaOptions

fileStream

foreachRDD

saveAs***Files

saveAs***Files

foreachRDD

dstream.foreachRDD{(rdd,time)=>
rdd.foreachPartition{partitionIterator=>
valpartitionId=TaskContext.get.partitionId()
valuniqueId=generateUniqueId(time.milliseconds,partitionId)
//usethisuniqueIdtotransactionallycommitthedatainpartitionIterator
}
}

StreamingContext.socketStream

FlumeUtils.createStream

stop

restart

org.apache.spark.streaming.receiver.Receiver

org.apache.spark.streaming.dstream.NetworkReceiver

onStart()

onStop()

store(<data>)

store(...)

org.apache.spark.streaming.receivers.Receiver

org.apache.spark.streaming.receiver.ActorHelper

pushBlock(...)

store(...)

org.apache.spark.streaming.receivers

org.apache.spark.streaming.receiver