spark基础知识之spark streaming<三>
spark streaming:
1.设置合理的batchDuration 每隔一个batchDuration提交一个Job，可以达到毫秒级别，最好不要小于500ms； 2.增Job的并行度 尽可能把Task分配到不同的节点上 3.使用Kryo的序列化 spark默认使用 Java 的序列化 4.缓存经常使用的数据 缓存经常使用的数据，加快数据的处理 5.清楚不需要的数据 随着时间的推移，有一些数据是不需要的，但这些数据是缓存在内存中的，会消耗内存资源， 通过配置spark.cleaner.ttl为一个合理的值(值不能设置太小), 还可以配置spark.streaming.unpersist为true来去除持久化RDD(默认是true)。 6.设置合理的GC GC比较难调，不合理的GC行为会给程序带来很大的影响。在集群环境下，可以使用并行Mark-Sweep垃圾回收机制(消耗的资源比较多)， 可以如下配置：spark.executor.extraJavaOptions=-XX:+UseConcMarkSweepGC 7.设置合理的CPU资源 在streaming情况下，很多时候需要的内存不是很多，但需要的CPU要很多。在streaming程序中，CPU资源的使用可以分为两大类：(1)、用于接受数据；(2)、用于处理数据。 sparkSQL 1.shuffle时候的Task数量，通过spark.sql.shuffle.partitions来调节(默认是200)。 2.想要获取更好的表达式查询速度，可以将spark.sql.codegen设置为ture。 3.对于数据倾斜，采用加入部分中间步骤，如聚合后cache。 4.适当的使用序化方案以及压缩方案。 spark webUI监控Stage的一个图： 1.PROCESS_LOCAL是指读取缓存在本地节点的数据。 2.NODE_LOCAL是指读取本地节点硬盘数据。 3.ANY是指读取非本地节点数据。 4.通常读取数据PROCESS_LOCAL>NODE_LOCAL>ANY，尽量使数据以PROCESS_LOCAL或NODE_LOCAL方式读取。其中PROCESS_LOCAL还和cache有关。 在Spark集群环境下，只有足够高的并行度才能使 系统 资源得到充分的利用，可以通过修改spark-env.sh来调整Executor的数量和使用资源，Standalone和YARN方式资源的调度管理是不同的。 在Standalone模式下: 1. 每个节点使用的最大内存数：SPARK_WORKER_INSTANCES*SPARK_WORKER_MEMORY； 2. 每个节点的最大并发task数：SPARK_WORKER_INSTANCES*SPARK_WORKER_CORES。 在YARN模式下： 1. 集群task并行度：SPARK_ EXECUTOR_INSTANCES* SPARK_EXECUTOR_CORES； 2. 集群内存总量：(executor个数) * (SPARK_EXECUTOR_MEMORY+ spark.yarn.executor.memoryOverhead)+ (SPARK_DRIVER_MEMORY+spark.yarn.driver.memoryOverhead)。 重点强调：Spark对Executor和Driver额外添加堆内存大小，Executor端：由 spark.yarn.executor.memoryOverhead设置，默认值executorMemory * 0.07与384的最大值。Driver端：由spark.yarn.driver.memoryOverhead设置，默认值driverMemory * 0.07与384的最大值。 通过调整上述参数，可以提高集群并行度，让系统同时执行的任务更多，那么对于相同的任务，并行度高了，可以减少轮询次数。举例说明：如果一个stage有100task，并行度为50，那么执行完这次任务，需要轮询两次才能完成，如果并行度为100，那么一次就可以了。 但是在资源相同的情况，并行度高了，相应的Executor内存就会减少，所以需要根据实际实况协调内存和core。此外，Spark能够非常有 效的支持短时间任务（例如：200ms），因为会对所有的任务复用JVM，这样能减小任务启动的消耗，Standalone模式下，core可以允许 1-2倍于物理core的数量进行超配。 Spark任务数量调整 Spark的任务数由stage中的起始的所有RDD的partition之和数量决定，所以需要了解每个RDD的partition的计算方 法。以Spark应用从HDFS读取数据为例，HadoopRDD的partition切分方法完全继承于MapReduce中的 FileInputFormat，具体的partition数量由HDFS的块大小、mapred.min.split.size的大小、文件的压缩方式 等多个因素决定，详情需要参见FileInputFormat的代码。 Spark内存调优 内存优化有三个方面的考虑：对象所占用的内存，访问对象的消耗以及垃圾回收所占用的开销。 1. 对象所占内存，优化数据结构 Spark 默认使用Java序列化对象，虽然Java对象的访问速度更快，但其占用的空间通常比其内部的属性数据大2-5倍。为了减少内存的使用，减少Java序列 化后的额外开销，下面列举一些Spark官网（http://spark.apache.org/docs/latest /tuning.html#tuning-data-structures）提供的方法。 （1）使用对象数组以及原始类型（primitive type）数组以替代Java或者Scala集合类（collection class)。fastutil 库为原始数据类型提供了非常方便的集合类，且兼容Java标准类库。 （2）尽可能地避免采用含有指针的嵌套数据结构来保存小对象。 （3）考虑采用数字ID或者枚举类型以便替代String类型的主键。 （4）如果内存少于32GB，设置JVM参数-XX:+UseCom-pressedOops以便将8字节指针修改成4字节。与此同时，在 Java 7或者更高版本，设置JVM参数-XX:+UseC-----ompressedStrings以便采用8比特来编码每一个ASCII字符。 2. 内存回收 （1）获取内存统计信息：优化内存前需要了解集群的内存回收频率、内存回收耗费时间等信息，可以在spark-env.sh中设置 SPARK_JAVA_OPTS=“-verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps $ SPARK_JAVA_OPTS”来获取每一次内存回收的信息。 （2）优化缓存大小：默认情况Spark采用运行内存（spark.executor.memory）的60%来进行RDD缓存。这表明在任务执 行期间，有40%的内存可以用来进行对象创建。如果任务运行速度变慢且JVM频繁进行内存回收，或者内存空间不足，那么降低缓存大小设置可以减少内存消 耗，可以降低spark.storage.memoryFraction的大小。 3. 频繁GC或者OOM 针对这种情况，首先要确定现象是发生在Driver端还是在Executor端，然后在分别处理。 Driver端：通常由于计算过大的结果集被回收到Driver端导致，需要调大Driver端的内存解决，或者进一步减少结果集的数量。 Executor端： （1）以外部数据作为输入的Stage：这类Stage中出现GC通常是因为在Map侧进行map-side-combine时，由于group 过多引起的。解决方法可以增加partition的数量（即task的数量）来减少每个task要处理的数据，来减少GC的可能性。 （2）以shuffle作为输入的Stage：这类Stage中出现GC的通常原因也是和shuffle有关，常见原因是某一个或多个group 的数据过多，也就是所谓的数据倾斜，最简单的办法就是增加shuffle的task数量，比如在SparkSQL中设置SET spark.sql.shuffle.partitions=400，如果调大shuffle的task无法解决问题，说明你的数据倾斜很严重，某一个 group的数据远远大于其他的group，需要你在业务逻辑上进行调整，预先针对较大的group做单独处理。 修改序列化 使用Kryo序列化，因为Kryo序列化结果比Java标准序列化更小，更快速。具体方法：spark-default.conf 里设置spark.serializer为org.apache.spark.serializer.KryoSerializer 。 参考官方文档（http://spark.apache.org/docs/latest/tuning.html#summary）：对于大多数程序而言，采用Kryo框架以及序列化能够解决性能相关的大部分问题。 Spark 磁盘调优 在集群环境下，如果数据分布不均匀，造成节点间任务分布不均匀，也会导致节点间源数据不必要的网络传输，从而大大影响系统性能，那么对于磁盘调优最好先将数据资源分布均匀。除此之外，还可以对源数据做一定的处理： 1. 在内存允许范围内，将频繁访问的文件或数据置于内存中； 2. 如果磁盘充裕，可以适当增加源数据在HDFS上的备份数以减少网络传输； 3. Spark支持多种文件格式及压缩方式，根据不同的应用环境进行合理的选择。如果每次计算只需要其中的某几列，可以使用列式文件格式，以减少磁盘I/O， 常用的列式有parquet、rcfile。如果文件过大，将原文件压缩可以减少磁盘I/O，例如：gzip、snappy、lzo。 其他 广播变量（broadcast） 当task中需要访问一个Driver端较大的数据时，可以通过使用SparkContext的广播变量来减小每一个任务的大小以及在集群中启动 作业的消耗。参考官方文档http://spark.apache.org/docs/latest /tuning.html#broadcasting-large-variables。 开启推测机制 推测机制后，如果集群中，某一台机器的几个task特别慢，推测机制会将任务分配到其他机器执行，最后Spark会选取最快的作为最终结果。 在spark-default.conf 中添加：spark.speculation true 推测机制与以下几个参数有关： 1. spark.speculation.interval 100：检测周期，单位毫秒； 2. spark.speculation.quantile 0.75：完成task的百分比时启动推测； 3. spark.speculation.multiplier 1.5：比其他的慢多少倍时启动推测。 spark.driver.extraJavaOptions -XX:PermSize=128M -XX:MaxPermSize=256M cluster模式： ? 1 2 3 4 5 spark.driver.extraJavaOptions -XX:PermSize=1024M -XX:MaxPermSize=2048M (可被此参数覆盖--driver-java-options) spark.executor.extraJavaOptions=-XX:+UseConcMarkSweepGC client模式： JAVA_OPTS= "-XX:MaxPermSize=256m $OUR_JAVA_OPTS" ? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 spark.driver.memory=1g spark.driver.cores= 1 spark.driver.maxResultSize=1g spark.executor.memory=1g spark.executor.cores spark.shuffle.manager=sort spark.shuffle.file.buffer=64K spark.reducer.maxSizeInFlight=96m spark.shuffle.io.maxRetries= 10 spark.shuffle.sort.bypassMergeThreshold= 200 spark.shuffle.spill.compress= false spark.shuffle.memoryFraction= 0.1 spark. default .parallelism= 100 spark.storage.memoryFraction= 0.3 spark.yarn.executor.memoryOverhead= 1024 spark.yarn.driver.memoryOverhead= 1228 spark.yarn.am.memory= 2048 spark.yarn.am.memoryOverhead= 512 spark.eventLog.enabled= true spark.cleaner.ttl= 500 参数名称 含义 ? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 --master MASTER_URL 可以是spark: //host:port, mesos://host:port, yarn, yarn-cluster,yarn-client, local --deploy-mode DEPLOY_MODE Driver程序运行的地方，client或者cluster -- class CLASS_NAME 主类名称，含包名 --name NAME Application名称 --jars JARS Driver依赖的第三方jar包 --py-files PY_FILES 用逗号隔开的放置在Python应用程序PYTHONPATH上的 .zip, .egg, .py文件列表 --files FILES 用逗号隔开的要放置在每个executor工作目录的文件列表 --properties-file FILE 设置应用程序属性的文件路径，默认是conf/spark-defaults.conf --driver-memory MEM Driver程序使用内存大小 --driver-java-options --driver-library-path Driver程序的库路径 --driver- class -path Driver程序的类路径 --executor-memory MEM executor内存大小，默认1G --driver-cores NUM Driver程序的使用CPU个数，仅限于Spark Alone模式 --supervise 失败后是否重启Driver，仅限于Spark Alone模式 --total-executor-cores NUM executor使用的总核数，仅限于Spark Alone、Spark on Mesos模式 --executor-cores NUM 每个executor使用的内核数，默认为 1 ，仅限于Spark on Yarn模式 --queue QUEUE_NAME 提交应用程序给哪个YARN的队列，默认是 default 队列，仅限于Spark on Yarn模式 --num-executors NUM 启动的executor数量，默认是 2 个，仅限于Spark on Yarn模式 --archives ARCHIVES 仅限于Spark on Yarn模式 --conf "spark.mongodb.input.uri=mongodb://127.0.0.1/test.myCollection?readPreference=primaryPreferred" \ --conf "spark.mongodb.output.uri=mongodb://127.0.0.1/test.myCollection" \ --packages org.mongodb.spark:mongo-spark-connector_2. 10 : 1.1 . 0 总结 Spark系统的性能调优是一个很复杂的过程，需要对Spark以及Hadoop有足够的知识储备。从业务应用平台（Spark）、存储 （HDFS）、操作系统、硬件等多个层面都会对性能产生很大的影响。借助于多种性能监控工具，我们可以很好地了解系统的性能表现，并根据上面介绍的经验进 行调整。