• Spark SQL 之 Join 实现: 包含sparksql总体流程 join的基本要素 join基本实现流程

• 删除非空数据库: drop database <DATABASE> if exists cascade

在字段类型是float以及double的时候对not-a-number (NaN)进行了特殊的处理

• NaN = NaN returns true.
• In aggregations, all NaN values are grouped together.
• NaN is treated as a normal value in join keys.
• NaN values go last when in ascending order, larger than any other numeric value.

spark查看执行计划和RDBMS一样,使用的是explain关键字.执行计划中经常看到的Tungsten是spark 1.4新的储存引擎,这里有关于,如果想要查看逻辑计划和查询优化器,使用explain extended

如果使用编程的方式,在spark1中使用sqlContext.sql("your SQL query").explain(true),在spark2中使用spark.sql("your SQL query").explain(true)

• distribute by rand(): 重新分区TungstenExchange hashpartitioning(_nondeterministic#xxx,<PARTITION_NUM>),其中_nondeterministic对应的rand(如果是具体的filed就是显示TungstenExchange hashpartitioning(filed#xxx,<PARTITION_NUM>)),PARTITION_NUM和set spark.shuffle.partitions=<PARTITION_NUM>,这个可以检查repartition是否和预想中的一样

• select field1, field2: 选择了字段Project [field1, field2]]

• from <TABLENAME>: 扫描数据

  • 如果是parquet类型的表: Scan ParquetRelation <DATABASE>.<TABLENAME>[field1, field2] InputPaths: hdfs://domain:port/path/to/TABLENAME_FOLDER
  • 如果是textfile类型的表: HiveTableScan [filed1, field2] MetastoreRelation <DATABASE>, <TABLENAME>, <NONE>, [(ds = <DS_CLAUSE>)],注意: HiveTableScan spark2.0以下的版本会扫描全部的字段,这是一个性能的bug,SPARK-18642已经修复了

• where <CONDITION>: 过滤条件Filter (field <= <CONDITION>)

• <TABLENAME1> join <TABLENME2> on <FIELD1> = <FIELD2>: 数据间的关联

  • sortMergeJoin: 包括sortMergeLeftJoin,一般的执行计划为

    +- Sort

• limit <LIMIT_NUM>: Limit <LIMIT_NUM>获取前几条

• 部分

缓存表,实现hive的with <table_name> as (<select clause>),两者都只对当前session有效,spark最好手动删除cache表,hive with clause会在当前session退出的时候自动注销with创建的表.

-- 缓存全表
cache table <real_table_name>;
-- 缓存结果表
cache table <cache_table_name> as select * from <real_table_name>;
-- 取消 cache 表
uncache table <real_table_name|cache_table_name>;

-- 当需要使用的时候再实现缓存的表 关键字 lazy 等到有action操作再缓存数据使用
cache lazy table <real_table_name>