王家林spark語(yǔ)錄第一季版

中使用--driver-java-options和使用spark.executor.extraJavaOptions參數(shù)設(shè)置減少Batch的處理時(shí)間。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0153（2016.4.18）:在SparkStreaming中強(qiáng)烈建議在Standalone模式下運(yùn)行，因?yàn)榭梢赃_(dá)到更短的Task啟動(dòng)時(shí)間。BatchBatchIntervalSparkStreamingTask個(gè)DStream，后續(xù)的Transformation算子操作都針對(duì)該一個(gè)聚合后的DStream即可。希望影響系統(tǒng)的吞吐量，那么可以創(chuàng)建多個(gè)輸入DStream，啟動(dòng)多個(gè)Rceiver。日志。這種機(jī)制可以讓driver中對(duì)Executor5為設(shè)置為滑動(dòng)創(chuàng)建大小的5到10倍的樣子。中如果想讓Driver從失敗中恢復(fù)回來(lái)，處理要指定checkpoint目錄并且使用一個(gè)StreamingContext實(shí)例。checkpointStatefulTransformationsCheckpoint能夠獲得主要價(jià)值在于對(duì)Stateful的操作失敗的時(shí)候進(jìn)行快速的恢復(fù)。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0136（2016.4.17）:在SparkStreaming中對(duì)于Window操作默認(rèn)就開(kāi)啟了持久化機(jī)制，例如reduceByWindow、開(kāi)發(fā)人員手動(dòng)調(diào)用persist方法。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0135（2016.4.17）:在SparkStreaming開(kāi)發(fā)人員可以手動(dòng)控制數(shù)據(jù)流的持久化，如果要對(duì)一個(gè)DStream進(jìn)行多次操作，那么對(duì)操作以提高性能，可以考慮使用SparkStreaming提供的mapPartitions方法。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0133（2016.4.16）:SparkStreamingtransformDStreamSparkSparkSQL、圖計(jì)算和機(jī)器學(xué)習(xí)功能等。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0132（2016.4.15）:SparkStreaming入到一個(gè)數(shù)據(jù)庫(kù)，需要特別注意的是foreachRDD函數(shù)是在Driver中執(zhí)行.王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0131（2016.4.15）:SparkStreamingDStream來(lái)從Kafka等數(shù)據(jù)源中讀取的一個(gè)輸入流。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0130（2016.4.15）:SparkStreaming中的inputDStreamDStreamSparkStreaming任何后續(xù)的DStream都會(huì)建立標(biāo)準(zhǔn)的DStreams。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0129（2016.4.14）:SparkKakafa王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0128（2016.4.14）:SparkStreaming王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0128（2016.4.14）:SparkStreaming王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0127（2016.4.13）:SparkStreaming通過(guò)創(chuàng)建線程池方式來(lái)接收Kafka中的Topics數(shù)據(jù)的。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0126（2016.4.13）:SparkStreaming和處理來(lái)自ReceiverTracker和分布式的Receivers發(fā)送的消息，例如處理StartAllReceivers、UpdateReceiverRateLimit等。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0125（2016.4.13于深圳）:SparkStreaming中用戶可以自由繼承ReceiverInputDStream并自定義相應(yīng)的Receiver，就是SparkStreaming能兼容眾多數(shù)據(jù)源的原因，例如對(duì)于KafkaInputDStream繼承了可以從Kafka獲取數(shù)據(jù)了。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0124（2016.4.13）:SparkStreamingJobs，doCheckpointcheckpoint。在doCheckpointStreamingApplicationCheckpointDuration就進(jìn)行Checkpoint操作。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0123（2016.4.13于深圳）:SparkStreamingStreamingApplicationCheckpointCheckpointCheckpointCheckPoint的時(shí)間設(shè)置為Batchduration510倍；王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0122（2016.4.13）:SparkStreamingupdateStateByKey、reduceByKeyAndWindowStateful須提供checkpoint目錄來(lái)允許定時(shí)的RDDcheckpoint操作。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0121（2016.4.13于深圳）:SparkStreamingCheckpointDataRDDStatefulRDD中間生成的RDDs保存到可靠存儲(chǔ)來(lái)切斷依賴鏈；王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0120（2016.4.12）:SparkStreamingCheckpointMetadatastreamingapplication的所有配置、王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0119（2016.4.12）:SparkStreamingReceiverSupervisorImpl在收到單塊條小數(shù)據(jù)后會(huì)委托BlockGenerator積累并封裝每收到一條ReceiverSupervisorImpl轉(zhuǎn)發(fā)來(lái)的數(shù)據(jù)就加入到這個(gè)currentBuffer數(shù)組王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0118（2016.4.12）:SparkStreamingReceivedBlockHandlerExecutorBlockManagerBasedBlockHandler，是直接存到Executor的內(nèi)存或硬盤王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0117（2016.4.12）:SparkStreamingExecutor中的ReceiverSupervisorImplstart()核心工作就是調(diào)用Receiver.onStart()，來(lái)啟動(dòng)Receiver的數(shù)據(jù)接收線程；王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0116（2016.4.12）:SparkStreamingDriver中的ReceiverTracker將收到的塊數(shù)據(jù)Meta信息直接轉(zhuǎn)給自己的成員王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0115（2016.4.12）:SparkStreaming中在Driver端通過(guò)ReceivedBlockTracker獲得元數(shù)據(jù)后就會(huì)由ReceiverInputDStream在BatchReceiverTrackerMetabatch內(nèi)處理，然后生成相應(yīng)的RDD實(shí)例去處理這些塊數(shù)據(jù)。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0114（2016.4.12）:SparkStreamingReceiver在onStart()啟動(dòng)后，就將持續(xù)不斷地接收外界數(shù)據(jù)，并持續(xù)交給ReceiverSupervisor進(jìn)行數(shù)據(jù)存儲(chǔ)并把數(shù)據(jù)的MetaDriver中的王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0113（2016.4.12）:SparkStreaming中Receiver的總指揮ReceiverTracker分發(fā)多個(gè)jobexecutor上分別啟動(dòng)產(chǎn)生或者持續(xù)接收系統(tǒng)外數(shù)據(jù)Receiver實(shí)現(xiàn)的實(shí)例并在Receiver實(shí)例生成后調(diào)用王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0112（2016.4.12）:SparkStreamingReceiver1TaskJobTask1ReceiverTask王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0111（2016.4.12）:SparkStreaming中一個(gè)Receiver的分發(fā)Job是有可能沒(méi)有完成分發(fā)Receiver的目的的，此時(shí)ReceiverTracker會(huì)繼續(xù)再起一個(gè)Job來(lái)嘗試Receiver分發(fā)，這就保證了如果一次SparkStreaming層面對(duì)Receiver所在位置更好的控制。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0110（2016.4.12）:SparkStreaming中的ReceiverJobReceiverJob王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0109（2016.4.12）:SparkStreamingSparkStreamingContext.start(ReceiverTracker.start()，而ReceiverTracker.start(launchReceiversReceiverexecutor上去，再然后在每個(gè)executor上由ReceiverSupervisor來(lái)分別啟動(dòng)一個(gè)Receiver接收數(shù)據(jù)。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0108（2016.4.12）:SparkStreaming中的ReceiverTrackerdriverReceiverTrackerMaster各個(gè)Executor上的Receivers。是由outputstream通過(guò)dependency引用關(guān)系索遞歸的追溯到最上游的InputDStream節(jié)點(diǎn)時(shí)，最上游的節(jié)點(diǎn)InputDStreamInputStream王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0106（2016.4.12）:SparkStreaming下JobGenerator中RecurringTimer的調(diào)度周期是batchDuration，每次調(diào)度都是往eventLoop里發(fā)送一個(gè)消息，eventLoopbatch(newTime(longTime))GenerateJobs。是真正調(diào)用了Job.run()方法觸發(fā)Job.func的真正執(zhí)行執(zhí)行的的時(shí)候是在名為在ForEachDStream.generateJob(timeJobJob.func，而在JobHandler中才是是真正調(diào)用了Job.run()方法觸發(fā)Job.func的真正執(zhí)行！Seq[Job]后會(huì)將其包裝成一個(gè)JobSet，然后轉(zhuǎn)過(guò)來(lái)調(diào)用JobScheduler.submitJobSet(jobSet)來(lái)交付回JobScheduler。jobSetjobjobExecutor.execute(newJobHandler(job))，其實(shí)質(zhì)是將每個(gè)job都在jobExecutor線程池中、用newJobHandler來(lái)處理。0102016.4.1于深圳:JobSchedule定時(shí)為每個(gè)batch生成RDDDAG的實(shí)例。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0100（2016.4.12）:SparkStreaming工作委托給JobGenerator，而將源頭輸入數(shù)據(jù)的記錄工作委托給ReceiverTracker。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0099（2016.4.12）:SparkStreaming中在給定的batch里面，DStreamGraph.generateJobs(time)的工作完成時(shí)候會(huì)產(chǎn)生Seq[Job]作為結(jié)果返回給JobGenerator，JobGenerator會(huì)把Seq[Job]盡快提交到王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0098（2016.4.12）:SparkStreaming中在給定的batch里面，DStreamGraph.generateJobs(time)的工作完成時(shí)候會(huì)產(chǎn)生Seq[Job]作為結(jié)果返回給JobGenerator，JobGenerator會(huì)把Seq[Job]盡快提交到JobShedulerJob.run()使得Seq[JobRDDDAG操作，這和Java中的Runnable接口很像.JobGenerator調(diào)用DStreamGraph的generateJobs(time)方法，而StreamGraph繼續(xù)調(diào)用了每個(gè)輸出流的generateJob(time產(chǎn)生的來(lái)源，在每個(gè)BatchRDD,能夠這樣做的根本原因是DStream的TransformationAPIRDDTransformationAPI中的)DStreamGraphInputStreamInputStream而對(duì)OutputStreamingBFS遍歷的時(shí)間消耗。中的DStreamTransformationDAG，在物理實(shí)現(xiàn)上是通過(guò)與Transformation操作反向的依賴來(lái)構(gòu)成表示的。王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0092（2016.4.12）:SparkStreamingoutputstreamForEachDStream王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0091（2016.4.12）:SparkStreamingExecutor對(duì)Kafka王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0090（2016.4.12于深圳）:SparkStreaming容錯(cuò)時(shí)候的熱備是指在存儲(chǔ)塊數(shù)據(jù)時(shí)在存儲(chǔ)到本地Executor的同時(shí)備份到另外一個(gè)現(xiàn)自己的Receiver時(shí)指定一下StorageLevel為MEMORY_ONLY_2或MEMORY_AND_DISK_2王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0089（2016.4.12）:SparkStreaming中的StreamingContextJobSchedulerJobGeneratorSparkStreaming0088（2016.4.12）:SparkStreaming王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0087（2016.4.12于深圳）:SparkStreamingSparkStreaming、GraphX、ML統(tǒng)一處理來(lái)自于Kafka、Flume等不用來(lái)源的數(shù)據(jù)；王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0086（2016.4.12）:SparkStreamingFlumeSpark所有其它功能無(wú)縫整合！王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0085（2016.4.12）:SparkStreamingKafkaSparkStreamingBatchComputingSparkSQLAPI由Dataset表示，而在Spark2.0Spark1.3引入的DataFrame則變成Dataset[Row]的類型別名；SparkStreamingSpark王家林每日大數(shù)據(jù)語(yǔ)錄SparkStreaming0082（2016.4.12）:SparkStreaming王家林每日大數(shù)據(jù)語(yǔ)錄Scala0081（2016.4.11）:Scala作用推到越晚越好的理念，而Spark正是這一理念的完美應(yīng)用！王家林每日大數(shù)據(jù)語(yǔ)錄Spark0077（2016.1.23）:SparkJob比較容易內(nèi)存溢出，另外一個(gè)比較有效的辦法是減少并行的Executor數(shù)量，這樣每個(gè)比較容易內(nèi)存溢出，一個(gè)比較有效的辦法就是增加Task的并行度，這樣每個(gè)Task處理的Partition的數(shù)據(jù)量就變少了，減少了OOM的可能性。某些TaskExecutor的個(gè)數(shù)，這樣每個(gè)Task王家林每日大數(shù)據(jù)語(yǔ)錄Spark0072（2016.1.22）:默認(rèn)情況下SparkExecutorCore，這樣帶來(lái)的一個(gè)好處就是可以最大化的提高計(jì)算存比較大，就需要頻繁的spillover或者有更多的OOM的風(fēng)險(xiǎn)。007（2016.1.:SparkWorker，WorkerExecutor下面又可以有若干個(gè)Executor。邏輯完全相同但處理數(shù)據(jù)不同的分布式并行運(yùn)行的Task構(gòu)成，Stage內(nèi)部的計(jì)算都以PipelineStageShuffle的唯一方式。王家林每日大數(shù)據(jù)語(yǔ)錄Spark0069（2016.1.21）:在SparkWorker節(jié)點(diǎn)上使用固態(tài)硬盤以及把WorkerShuffleRAMDiskShuffleSortBasedShuffleShuffleReaderHashShuffleReader。實(shí)際上是MapStatus，在DAGScheduler的handleTaskCompletion中會(huì)通過(guò)上一個(gè)Stage的數(shù)據(jù)了。王家林每日大數(shù)據(jù)語(yǔ)錄Spark0066（2016.1.21）:TaskRunnerTaskExecutorBackendstatusUpdateDriverDriver會(huì)轉(zhuǎn)發(fā)給自己的Shuffle數(shù)據(jù)。王家林每日大數(shù)據(jù)語(yǔ)錄Spark0064（2016.1.20）Shuffle別適合于集群中有大量?jī)?nèi)存以及有很多計(jì)算任務(wù)的情景，通過(guò)Tachyon可以使得多個(gè)新去計(jì)算丟失的Partition的數(shù)據(jù)。王家林每日大數(shù)據(jù)語(yǔ)錄Spark0061（2016.1.19于深圳）:Spark的計(jì)算盡量不要落到磁Partition該實(shí)例的引用句柄并且通過(guò)該句柄來(lái)維持Executor和Driver之間的心跳。理工作且響應(yīng)Master獲取Block的狀態(tài)的請(qǐng)求。式spark.local.dir或者通過(guò)SPARK_LOCAL_DIRS設(shè)置的目錄中。王家林每日大數(shù)據(jù)語(yǔ)錄Spark0056（2016.1.18于深圳）:所有的Executor中的己，注冊(cè)的信息包括blockManagerId，當(dāng)前Node可用的最大內(nèi)存數(shù)據(jù)以及當(dāng)前BlockManagerSlaveActor的引用等。DAGScheduler中就是通過(guò)這些信息來(lái)確認(rèn)數(shù)據(jù)運(yùn)行的本地性的！化是通過(guò)Storage模塊來(lái)完成的。志中看出每個(gè)Cache分區(qū)的大小的估計(jì)值。王家林每日大數(shù)據(jù)語(yǔ)錄Spark0051（2015.1.13）:TaskRunnerTask狀態(tài)匯報(bào)給SchedulerBackend的DriverActorDriverActor會(huì)把信息轉(zhuǎn)發(fā)給TaskScheduler的statusUpdate進(jìn)行處理。王家林每日大數(shù)據(jù)語(yǔ)錄Spark0050（2015.12.30）:SparkTask從序列化的TaskTask執(zhí)行時(shí)依賴的文件和文件下載的位置信息，然后使用王家林每日大數(shù)據(jù)語(yǔ)錄Spark0049（2015.12.29于深圳）:SparkMaster在為的計(jì)算盡可能的分配到不同的計(jì)算節(jié)點(diǎn)上，這也是Spark默認(rèn)的方式；另外一種是給004（2015.12.29:SparkDploySchdulerBakendexecutorAdded、executorRemoved等事件并進(jìn)行相關(guān)事件的處理。前的Application分配Executor，最終在通過(guò)Executor并發(fā)的執(zhí)行JobTasksCuratorAPIZooKeeper。來(lái)的啟動(dòng)ExecutorExecutorRunner，借助ExecutorRunner的fetchAndRunExecutor會(huì)最終創(chuàng)建出ExecutorBackend進(jìn)程。Task的最終計(jì)算。王家林每日大數(shù)據(jù)語(yǔ)錄Spark0043（2015.12.15）:Worker最終AppClientSparkDeploySchedulerBackendMasterStatusUpdate信息來(lái)處理Executor丟失的信息，Task會(huì)被重新分配。王家林每日大數(shù)據(jù)語(yǔ)錄Spark0042（2015.12.15）SparkMaster般選擇使用ZooKeeperHA,如果當(dāng)前管理集群的Master失敗，ZooKeeper會(huì)在StandbyApplicationMasterACTIVE1665-151030-90Master發(fā)送消息。王家林每日大數(shù)據(jù)語(yǔ)錄Spark0038（2015.12.13于上海）:SparkMaster實(shí)現(xiàn)基于ZooKeeperHAZooKeeperMaster王家林每日大數(shù)據(jù)語(yǔ)錄Spark0037（2015.12.12）:SparkMaster接受MasterExecutorExecutor成Task的計(jì)算。王家林每日大數(shù)據(jù)語(yǔ)錄Spark0035（2015.12.11）:SparkonMesos王家林每日大數(shù)據(jù)語(yǔ)錄Spark0034（2015.12.11）:SparkonMesos王家林每日大數(shù)據(jù)語(yǔ)錄Spark0033（2015.12.2）:SparkClusterManagerStandalone、YARN、Mesos、EC2、LocalStandalone展了Spark的應(yīng)用場(chǎng)景、同時(shí)也極大的降低了部署和使用Spark的難度。過(guò)程調(diào)用TaskRunner的run方法。要發(fā)送Shuffle或者生成最終的計(jì)算結(jié)果。王家林每日大數(shù)據(jù)語(yǔ)錄Spark0030（2015.11.25）:TaskSparkPartitonSpapartitionerbucketResultTask王家林每日大數(shù)據(jù)語(yǔ)錄Spark0029（2015.11.25）:SparkStandalone式而言，在創(chuàng)建SparkContext的時(shí)候會(huì)通過(guò)Master為用戶提交的計(jì)算分配計(jì)算資源TaskTaskSetManager的失敗，如果是其它的情況，修改spark.task.maxFailures來(lái)設(shè)置這個(gè)最大重試次數(shù)。TaskScheduler的具體是閑著statusUpdate來(lái)處理結(jié)果信息。了一組由Pool構(gòu)建的調(diào)度樹(shù)，這棵樹(shù)的葉子節(jié)點(diǎn)是TaskSetManager.了一組由Pool構(gòu)建的調(diào)度樹(shù)，這棵樹(shù)的葉子節(jié)點(diǎn)是TaskSetManager.王家林每日大數(shù)據(jù)語(yǔ)錄Spark篇0025（2015.11.24于上海）:Spark的內(nèi)部Task處理完數(shù)據(jù)后所保存的數(shù)據(jù)的位置信息，進(jìn)而獲取所依賴Stage的數(shù)據(jù)。上啟動(dòng)該Task，最終完成計(jì)算的調(diào)度過(guò)程。都是TaskSchedulerImpl，YarnCluster和YarnClient的TaskScheduler是YarnClusterScheduler和YarnClientClusterScheduler都是繼承自寫入DiskSparklineagelineageLineage，就會(huì)減少開(kāi)銷。Checkpoint：1.DAG賴上做Checkpoint，這個(gè)時(shí)候就可以避免應(yīng)為L(zhǎng)ineage重新計(jì)算而帶來(lái)的冗余計(jì)算。框架幫助我們隱式生成的，例如reduceByKey操作時(shí)候的