#!/bin/bash #队列名 根据yarn的队列提交 realtime_queue=root #提交的任务名 my_job_name="OrderQZ" spark-shell --master yarn --deploy-mode client --queue $realtime_queue #总的executors数 根据数据量与自己的集群资源来分配 --num-executors 35 #每个executor的核数 --executor-cores 5 #每个executor的内存 --executor-memory 19G #diver 端jvm日志配置 --conf spark.driver.extraJavaOptions=-Dlog4j.configuration=log4j-yarn.properties --conf spark.executor.extraJavaOptions=-Dlog4j.configuration=log4j-yarn.properties #序列化 --conf spark.serializer=org.apache.spark.serializer.KryoSerializer #数据本地化;一般会默认3s,重试5次的去分配，一旦超时失败，将会选择一个比上一个本地级别差的级别再一次分配，如果发生了数据传输，那么task首先通过blockmanager获取数据，如果本地没有数据，则通过getRemote方法从数据所在节点的blockmanager获取数据并返回至task所在节点 --conf spark.locality.wait=5 #失败重试次数 --conf spark.task.maxFailures=8 # 是否开启在webui杀死进程 --conf spark.ui.killEnabled=false #SparkContext 启动时记录有效 SparkConf信息 --conf spark.logConf=true #driver的堆外内存 内存对象分配在Java虚拟机的堆以外的内存，这些内存直接受操作系统管理（而不是虚拟机），这样做的结果就是能够在一定程度上减少垃圾回收对应用程序造成的影响。使用未公开的Unsafe和NIO包下ByteBuffer来创建堆外内存 --conf spark.yarn.driver.memoryOverhead=512 --conf spark.yarn.executor.memoryOverhead=5480 #提交申请的最大尝试次数, 小于等于YARN配置中的全局最大尝试次数。 --conf spark.yarn.maxAppAttempts=4 #定义AM故障跟踪的有效时间间隔。如果AM至少在定义的时间间隔内运行，则AM故障计数将被重置。如果未配置，此功能未启用。 --conf spark.yarn.am.attemptFailuresValidityInterval=1h --conf spark.yarn.executor.failuresValidityInterval=1h #动态资源分配 --conf spark.shuffle.service.enabled=true --conf spark.serializer=org.apache.spark.serializer.KryoSerializer #推测执行 --conf spark.speculation=true --conf spark.speculation.quantile=0.9 #shffule task数目 --conf spark.sql.shuffle.partitions=178 #，当它设置为true时，Spark SQL会把每条查询的语句在运行时编译为java的二进制代码。这有什么作用呢？它可以提高大型查询的性能，但是如果进行小规模的查询的时候反而会变慢，就是说直接用查询反而比将它编译成为java的二进制代码快。所以在优化这个选项的时候要视情况而定。 --conf spark.sql.codegen=true #默认值为false 它的作用是自动对内存中的列式存储进行压缩 --conf spark.sql.inMemoryColumnarStorage.compressed=true # join实现主要有3种，即BroadcastHashJoinExec、ShuffledHashJoinExec和SortMergeJoinExec，优先级为 #1 如果canBroadcast，则BroadcastHashJoinExec； #2 如果spark.sql.join.preferSortMergeJoin=false，则ShuffledHashJoinExec； #3 否则为SortMergeJoinExec； --conf spark.sql.join.preferSortMergeJoin=true # Spark底层shuffle的传输方式是使用netty传输，netty在进行网络传输的过程会申请堆外内存（netty是零拷贝），所以使用了堆外内存。 --conf spark.reducer.maxSizeInFlight=96M /** spark.reducer.maxSizeInFlight默认值：48m 参数说明：该参数用于设置shuffle read task的buffer缓冲大小，而这个buffer缓冲决定了每次能够拉取多少数据。 调优建议：如果作业可用的内存资源较为充足的话，可以适当增加这个参数的大小（比如96m），从而减少拉取数据的次数，也就可以减少网络传输的次数，进而提升性能。在实践中发现，合理调节该参数，性能会有1%~5%的提升。 错误：reduce oom reduce task去map拉数据，reduce 一边拉数据一边聚合 reduce段有一块聚合内存（executor memory * 0.2） 解决办法：1、增加reduce 聚合的内存的比例 设置spark.shuffle.memoryFraction 2、 增加executor memory的大小 --executor-memory 5G 3、减少reduce task每次拉取的数据量 设置spark.reducer.maxSizeInFlight 24m */