把项目中的注释和笔记整理了一下
（实际业务的变量名和列名肯定不是下面示例这样的......）
1. Spark的压缩
一般不用动。默认shuffle等阶段都会用压缩，看到有文章对比，Spark默认的lz4算法，压缩和解压时间略好于snappy。
2. spark读取多个嵌套文件夹
val myrdd = sc.textFile("data/*/part-*")
Spark天然就支持。RDD的binaryFiles和textfile、wholeTextFiles等方法都支持。
3. 分布式spark环境下读local文件
- 在分布式环境下读写local文件，发现结果集数量变少，但没有报错。但网上有文章说应该在读文件时报错，除非每个节点都存在数据集。
不报错的原因可能是设置了：
val spark = SparkSession.builder.appName("myApp")
.config("spark.files.ignoreCorruptFiles", "true")
.config("spark.sql.files.ignoreCorruptFiles", "true")
.getOrCreate
但理论上这只是过滤错误文件，也有点说不通，详细原因没有深究。
4. 利用SparkSession方法读CSV
val df = spark.read
.option("header", "true")//含表头
.option("delimiter", ";")//分隔符
.option("encoding", "GBK")//编码方式
.option("inferSchema","true") //一般建议关闭，性能有影响
.option("mode","DROPMALFORMED")//从数据集中删除损坏的记录
.csv("实际路径")
- 此时最好设置spark.sql.files.ignoreCorruptFiles。如果利用RDD方法读取csv（其实就是读文本，自己再切分）则不必设置这一行（只设置spark.files.ignoreCorruptFiles）。
- SparkSession没有binaryFiles的类似方法。
5. RDD转Dataframe
RDD转成Dataframe之后，代码好写很多，还可以用SQL，对比测试发现一般情况下效率和直接用RDD差不多。我用的方法是将数据（指定的列）转成list映射到case class，然后toDF
case class myschema(
col1: String,
col2: String,
col3: Double)
然后RDD端：
val mydf = myrdd
.map(x => x.split(","))
.map(x => List(x(0), x(5)，x(7))
.map(x => myschema(x(0), x(1), x(5).toDouble))
.toDF()
默认情况下会报错：
Lvalue toDF is not a member of org.apache.spark.myrdd，
解决方法是：
- import spark.sqlContext.implicits._
（注意还有个 spark.implicits._ 和 spark.sqlContext.implicits._不可同时存在） - case class 需要放在toDF 方法的作用域之外。
6. RDD和Dataframe的缓存
- 首先StorageLevel需要引用:
import org.apache.spark.storage.StorageLevel
//之后可以使用
myrdd.persist(StorageLevel.MEMORY_ONLY)
//或者
myrdd.cache()
- cache方法是一个Tranformation，不是action。此外persist是lazy级别的，unpersist时eager级别的。
- 对于RDD来说，cache方法相当于MEMORY_ONLY
- dataframe的cache方法相当于MEMORY_AND_DISK。MEMORY_AND_DISK也是dataframe的默认存储级别。
- 其他的内存级别：
- MEMORY_ONLY_2：保存两份
- MEMORY_ONLY_SER：数据序列化，降低内存溢出风险，但序列化有开销
- OFF_HEAP ：堆外内存，号称能减少内存消耗，减少GC，早期配合tachyon使用（参考：//www.greatytc.com/p/c6f6d4071560）
7. spark的重分区：repartition和coalesce
- 可以理解为：
- repartition()多用于增加分区（宽依赖，有shuflle）
- coalesce()多用于减少分区（窄依赖，没有shuffle）
- repartition方法是coalesce方法shuffle为true的情况
- 在使用filter之后进行coalesce效果比较明显。
8. 使用dataframe进行聚合统计
应该比较好看懂：
val mydf2 = mydf
.groupBy("col1", "col2")
.agg( "col3" -> "first","col4" -> "mean", "col5" -> "count")
//聚合之后的列名为: first(col3 ) avg(col4) count(col5)
.select(
bround($"avg(col4)", scale = 3).as("avgcol4"),
($"col2"),
($"first(col3)").as("fcol3"),
($"col1").as("col1"),
($"count(col5)").as("ccol5"))
最后发现还是用sql比较方便，因为更通用、更熟悉。比如上面描述的聚合之后的列名，用“$”符号表示当前表等语法，之前没怎么用过，查了资料才写对。
mydf.createTempView("mytable")
val df2 = spark.sql("""
select col1,
first(col3) as col3,
count(col4) as col4,
cast(avg(col5) - 140 as decimal(10,3)) as col5
from mytable
group by col1""")
这样比较方便，效率差不多，但貌似spark-sql要进行优化的话，方式语法和直接用dataframe有一些差别。
- 注意：
- 代码中的spark是SparkSession
- 代码是示随便写的，上下没严格对应关系
- 注意建表语句（createTempView）和sql语句的表名一致
- 输出的df2是个dataframe
- 用三引号是为了字符串换行
- 中文字段名要加单引号： first('字段1') as col1,
9. 在sql或dataframe中使用自定义函数
两种情况下自定义函数的写法是不同的。
spark-sql使用，需要将这个函数进行注册（register）：
import org.apache.spark.sql.functions.udf
spark.udf.register( "funname", ((x: Double, y: Double) => { x + y }))
val df2= spark.sql(" select funname(col1,col2) from table1")
dafaframe中使用：
val funname = udf((x: Double, y: Double) => { x + y })
val df2 = mydf.select(funname($"col1",$"col2"))
注意实际使用时要根据实际情况转换字段类型。
10. RDD去掉前n行数据
myrdd.zipWithIndex().filter(_._2>=n).keys
- zipWithIndex函数将RDD中的元素和元素的索引组成键值对。
(对比：zipWithUniqueId将RDD中元素和一个唯一ID组成键值对。) - 另一个方式：
val arr =df1.take(2)
val df2 =df1.filter(!arr.contains(_))
- 并没有进行认真的测试，最后业务中回避了这个逻辑。总是主观感觉在处理海量小文件的时候可能有性能问题。
