一 spark RDD分区原则
二.coalesce和repartition partitionby(后面一点说)
类别 | 函数 | 区别 |
---|---|---|
Transformation | Map,filter,groupBy,join, union,reduce,sort,partitionBy | 返回值还是 RDD ,不会马上 提交 Spark 集群运行 |
Action | count,collect,take,save, show | 返回值不是 RDD ,会形成 DAG 图,提交 Spark 集群运行 并立即返回结果 |
Transformation 操作不是马上提交 Spark 集群执行的,Spark 在遇到 Transformation 操作时只会记录需要这样的操作,并不会去执行,需要等到有 Action 操作的时候才会真正启动计算过程进行计算.针对每个 Action,Spark 会生成一个 Job, 从数据的创建开始,经过 Transformation, 结尾是 Action 操作.这些操作对应形成一个有向无环图(DAG),形成 DAG 的先决条件是最后的函数操作是一个Action.
所以建立索引文件,来标记shuffleMapTask在shuffleBlockFile的位置+偏移量,这样就可以在一个文件里面把不同的shuffleMaptask数据分出来
都是对数据进行重新分区,分区算法默认都是使用 HashPartitioner
七. DF的几种save Table 方式的比较:
7.1 df.write.mode(SaveMode.Append).saveAsTable("testDB.testTableName")
可以直接将df 保存在一个新的hive表中(不分区不需要事先在hive用语句建表),df中有多少分区,hive中就有多少分区,没有分区key
7.2 df.repartition(n).write.mode(SaveMode.Append).saveAsTable("testDB.testTableName")
df.repartition(n).write.mode(SaveMode.Overwrite ).saveAsTable("testDB.testTableName") 可以直接将df 保存在一个新的hive表中(因为这种方式的RDD的分区的key是随机生成的,所以保存到hive表中时并不会真正分区,所以也不需要事先在hive用语句建表),且 对df 重新分区,分区数为n,各个分区的key 随机生成。
注意:SaveMode.Overwrite 写入模式:不支持partitionby即
df.write.mode(SaveMode.Overwrite).partitionby("year","month","day").saveAsTable("testDB.testTableName") 会报错,即使事先在hive用语句建了分区表也不行,因为Overwrite 会先删除hive分区表,再重新建表,重新建表的时候不能自动建分区表,就会报错
7.3 df.repartition(n).write.mode(SaveMode.Append).partitionby("year","month","day").saveAsTable("testDB.testTableName")
可以将df 保存在一个新的hive表中(但是需要事先在hive用语句建分区表,根据partitionby的参数分区,为什么要这样做呢?DATAFRAM.SaveTestRead创建RDD分区,而不是HiVE分区。解决方法是在调用DataFrame.saveAsTable之前通过HQL创建表。),该语句会 对df 重新分区,分区数为n,各个分区的key 随机生成,然后再按年月日分区。即 把df 的数据 先按HashPartitioner算法 ,分布在n个分区中,再对每一个分区 按年月日分为 更小的分区。
7.4 df.write.mode(SaveMode.Append).partitionby("year","month","day").saveAsTable("testDB.testTableName")
可以将df 保存在一个新的hive表中(但是需要事先在hive用语句建分区表,根据partitionby的参数分区) 。即把df 的数据 按HashPartitioner算法 按年月日分为 更小的分区。df中有多少天的数据就有多少个分区。
想避免小文件。如果不重新分区的话,SparkSQL DataFrame保存到hdfs时出现过多小文件问题。
save Table 效率排名(分区数越多,效率越高,速度越快 ):
在n<天数的情况下 7.3>7.4>7.2 和7.1
解析:先将df 的小文件合并(一般df都是某个逻辑之后的结果,肯定有很多分区内的小文件),合并为1个分区能最大限度的合并小文件,spark1.6中用 coalesce(1)而不用repartition(1)的原因是,前者不需要进行shuffle操作,节约运行时间,最后再将这一个分区 按年月日 分为 多个小的分区,有多少天 就有多少个分区。(注意:需要事先在hive用语句建分区表,根据partitionby的参数分区;因为这里的分区只是RDD的分区,而不能设置hive分区) 。 最后为什么要按年月日分区呢,一方面把分区数从1再次增大,RDD分区数越多 ,可以提高程序运行(save)的并行度;另一方面 保存到分区表之后,再通过年月日 来筛选过滤数据,以进行某些查询(逻辑),会大大提高效率,因为同一天的数据放在同一分区,同一分区的数据 在物理地址上相邻(基本保持顺序存储),查询速度会更快。
八.Spark的并行度指的是什么?
2、如何设置一个Spark Application的并行度?
spark.defalut.parallelism 默认是没有值的,如果设置了值比如说10,是在shuffle的过程才会起作用(val rdd2 = rdd1.reduceByKey(_+_) //rdd2的分区数就是10,rdd1的分区数不受这个参数的影响)
// 如果提交脚本中设置了20个executor ,每个executor 有4个core,那么就设置成20*4 的2到3倍
new SparkConf().set("spark.defalut.parallelism",''160'')//
.set("spark.executor.memory",''15g'')
九. spark 提交语句
xx.sh
spark-submit --master yarn-client --class com.gree.test.Test --driver-memory 4g --num-executors 20 --executor-cores 4 --executor-memory 15g --driver-java-options '-Xms5g -Xmx5g -XX:+UseCompressedOops' --conf spark.driver.maxResultSize=2g --conf spark.executor.heartbeatInterval=15s --conf spark.yarn.executor.memoryOverhead=3072 --conf spark.memory.useLegacyMode=true --conf spark.shuff le.memoryFraction=0.5 --conf spark.storage.memoryFraction=0.3 --conf "spark.executor.extraJavaOptions=-XX:+UseG1GC -xx:InitiatingHeapOccupancyPercent=25 -XX:+PrintGCDetails -XX:+PrintGCTimeStamps" --queue production/home/..../mysubmit/xxx.jar
提交参数说明:
--class
: The entry point for your application (e.g.org.apache.spark.examples.SparkPi
)--master
: The master URL for the cluster (e.g.spark://23.195.26.187:7077
)--deploy-mode
: Whether to deploy your driver on the worker nodes (cluster
) or locally as an external client (client
) (default:client
) †--conf
: Arbitrary Spark configuration property in key=value format. For values that contain spaces wrap “key=value” in quotes (as shown).application-jar
: Path to a bundled jar including your application and all dependencies. The URL must be globally visible inside of your cluster, for instance, anhdfs://
path or afile://
path that is present on all nodes.application-arguments
: Arguments passed to the main method of your main class, if any
详见:https://blog.csdn.net/stardhb/article/details/50454680
在spark scala 里面设置:
val conf =new SparkConf().setAppName("testProject").setMaster("yarn-client")
//.setMaster("spark://bakmaster:7077")//standalong
//.setMaster("local[*]")
//.setMaster("local[6]")
//.setMaster("local")
//.setMaster("yarn-cluster")//集群的客户端模式
//.setMaster("mesos://HOST:5050")//集群模式
// mysql配置
conf.set("spark.driver.extraClassPath","/opt/cm-5.7.1/share/cmf/lib/mysql-connector-java-5.1.33-bin.jar")
conf.set("spark.executor.extraClassPath","/opt/cm-5.7.1/share/cmf/lib/mysql-connector-java-5.1.33-bin.jar")
// sqlserver配置
conf.set("spark.driver.extraClassPath","/opt/cloudera-manager/cm-5.7.1/share/cmf/lib/sqljdbc4.jar")
conf.set("spark.executor.extraClassPath","/opt/cloudera-manager/cm-5.7.1/share/cmf/lib/sqljdbc4.jar")
// oracle 配置
conf.set("spark.driver.extraClassPath","/opt/cloudera-manager/cm-5.7.1/share/cmf/lib/ojdbc6.jar")
conf.set("spark.executor.extraClassPath","/opt/cloudera-manager/cm-5.7.1/share/cmf/lib/ojdbc6.jar")
/*conf.set("spark.locality.wait.process","300s")
conf.set("spark.locality.wait.node","200s")
conf.set("spark.default.parallelism","300")
conf.set("spark.executor.memory","20g")
conf.set("spark.cores.max","100")
*/