把项目中的注释和笔记整理了一下
（实际业务的变量名和列名肯定不是下面示例这样的......）

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使用（参考：https://www.jianshu.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(_))

并没有进行认真的测试，最后业务中回避了这个逻辑。总是主观感觉在处理海量小文件的时候可能有性能问题。

Spark的零散笔记（一）