RDD中的cache
调用cache方法
val testRDD = sc.parallelize(Seq(elementA, elementB, elementC)).map(x => (x._1, x._2)).setName("testRDD")testRDD.cache()
dataframe中的cache
利用catalog以表的形式对数据进行缓存
import org.apache.spark.SparkConf
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.functions.udfval conf = new SparkConf().setAppName(s"test_app")
val spark = SparkSession.builder().config(conf).getOrCreate()
spark.read.parquet(s"${BASEPATH}/dws_live_mid_stat_order_di/event_day=${event_day}").createOrReplaceTempView(s"dwd_flow_sessionid_di")spark.catalog.cacheTable("dwd_flow_sessionid_di")
spark.catalog.uncacheTable("dwd_flow_sessionid_di")
SQL中的cache
spark.sql(s"""|cache table flow_basic_tmp as|select| *|from| test.tmp_live_mid_stat_order_di|""".stripMargin)
UNCACHE TABLE [ IF EXISTS ] table_identifier
SQL cache 相关文档,可以懒加载 CACHE TABLE - Spark 3.5.0 Documentation
需要注意的点 :
Spark.createDateFrame需要执行
基于DF 需要执行action,才能让RDD cache住