SparkStreaming性能調優大全！

SparkStreaming性能調優大全！

1、日誌已滿：

 spark.executor.logs.rolling.maxSize 

下面三個日誌rolling參數記得設置： 

spark.executor.logs.rolling.strategy size 

spark.executor.logs.rolling.maxSize 134217728 #default byte 

spark.executor.logs.rolling.maxRetainedFiles

下面是spark1.6的源碼：

[spark] RollingFileAppender {
  = = = = = = (* ).toString
  = =

 

2、Spark Streamingz對Kafka的Offset進行管理

zookeeper.session.timeout.ms

通常跳大3~5倍。

http://geeks.aretotally.in/spark-streaming-kafka-direct-api-store-offsets-in-zk/

http://www.tuicool.com/articles/vaUzquJ

[spark] SparkCuratorUtil Logging {

  = = = = (
      conf: SparkConfzkUrlConf: = ): CuratorFramework = {
    ZK_URL = conf.get(zkUrlConf)
    zk = CuratorFrameworkFactory.newClient(ZK_URLExponentialBackoffRetry())
    zk.start()
    zk
  }

3、 spark.task.maxFailures 

默認4，調整10左右

TaskSetManagerSuite SparkFunSuite LocalSparkContext Logging {
  TaskLocality.{}

  = SparkConf

  = .getTimeAsMs()
  = () {
    .beforeEach()
    FakeRackUtil.cleanUp()
  }

  test() {
    sc = SparkContext()
    sched = FakeTaskScheduler(sc())
    taskSet = FakeTask.createTaskSet()
    clock = ManualClock
    manager = TaskSetManager(schedtaskSetclock)

4、spark.streaming.kafka.maxRetries

默認1，調成3或者5

5、Spark Streaming鏈接Kafka用Direct方式。

6、怎麼調優？入口在哪？

答案就是Spark配置參數的地方：

1. $SPARK_HOME/conf/spark-env.sh  腳本上配置。 配置格式以下：

export SPARK_DAEMON_MEMORY=1024m

2. 編程的方式（程序中在建立SparkContext以前，使用System.setProperty（「xx」，「xxx」）語句設置相應系統屬性值)，

val conf = new SparkConf()

             .setMaster("local")

             .setAppName("CountingSheep")

             .set("spark.executor.memory", "1g")

val sc = new SparkContext(conf)

三、即在spark-shell下和spark-submit下配置

如：Scala> System.setProperty("spark.akka.frameSize","10240m")

System.setProperty("spark.rpc.askTimeout","800")

./bin/spark-submit --name "My app"

                            --master local[4] 

                            --conf spark.shuffle.spill=false

                            --conf "spark.executor.extraJavaOptions=-XX:+PrintGCDetails 

                                       -XX:+PrintGCTimeStamps"

                            myApp.jar 

spark-submit也會從默認配置文件conf/spark-defaults.conf裏選取配置項，格式以下：

spark.master            spark://iteblog.com:7077

spark.executor.memory   512m

spark.eventLog.enabled  true

spark.serializer        org.apache.spark.serializer.KryoSerializer

（一）環境變量spark-env.sh配置項

SCALA_HOME              #指向你的scala安裝路徑

MESOS_NATIVE_LIBRARY   #若是你要在Mesos上運行集羣的話

SPARK_WORKER_MEMORY  #做業可以使用的內存容量，默認格式1000M或者 2G (默認:  全部RAM去掉給操做系統用的1 GB);每一個做業獨立的內存空間由SPARK_MEM決定。

SPARK_JAVA_OPTS   #添加JVM選項。你能夠經過-D來獲取任何系統屬性 

eg: SPARK_JAVA_OPTS+="-Dspark.kryoserializer.buffer.mb=1024"

SPARK_MEM     #設置每一個節點所能使用的內存總量。他們應該和JVM‘s -Xmx選項的格式保持一致（e.g.300m或1g）。注意：這個選項將很快被棄用支持系統屬性spark.executor.memory，因此咱們推薦將它使用在新代碼中。

SPARK_DAEMON_MEMORY   #分配給Spark master和worker守護進程的內存空間(默認512M)

SPARK_DAEMON_JAVA_OPTS  #Spark master和worker守護進程的JVM選項（默認：none)

（二）System Properties

• spark.akka.frameSize: 控制Spark中通訊消息的最大容量 （如 task 的輸出結果），默認爲10M。當處理大數據時，task 的輸出可能會大於這個值，須要根據實際數據設置一個更高的值。若是是這個值不夠大而產生的錯誤，能夠從 worker的日誌 中進行排查。一般 worker 上的任務失敗後，master 的運行日誌上出現」Lost TID: 「的提示，可經過查看失敗的 worker 的日誌文件($SPARK_HOME/worker/下面的log文件) 中記錄的任務的 Serialized size of result 是否超過10M來肯定。

• spark.default.parallelism: 控制Spark中的分佈式shuffle過程默認使用的task數量，默認爲8個。若是不作調整，數據量大時，就容易運行時間很長，甚至是出Exception，由於8個task沒法handle那麼多的數據。 注意這個值也不是說設置得越大越好。

• spark.local.dir：Spark 運行時的臨時目錄，例如 map 的輸出文件，保存在磁盤的 RDD 等都保存在這裏。默認是 /tmp 這個目錄，而一開始咱們搭建的小集羣上 /tmp 這個目錄的空間只有2G，大數據量跑起來就出 Exception （」No space left on device」）了。

如何如何查看已配置好並生效的參數？

經過webui來進行查看，http://master:4040/