spark如何配置checkpoint

1、sparkSession配置checkpoint的方法

# step1: 在conf中添加checkpoint的保存地址
val spark = SparkSession.builder
      .appName(JobRegister.getJobName("xxx", s"xxxx"))
      .config("hive.exec.dynamic.partition", "true")
      .config("hive.exec.dynamic.partition.mode", "nonstrict")
      .config("spark.sql.broadcastTimeout", 3000)
      .config("spark.sql.sources.partitionOverwriteMode", "dynamic")
      .config("spark.checkpoint.dir", "/user/vc/projects_prod/checkpoint/data") # 配置路径
      .enableHiveSupport()
      .getOrCreate()

# step2: 保险期间，在引用spark时，再设置下：
spark.sparkContext.setCheckpointDir("/user/vc/projects_prod/checkpoint/data")

# step3: 在代码中添加checkpoint函数
val risk_msg = spark.sql(s"select * from temp.dwd_im_basic_info ")
      .cache()
      .checkpoint()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

2、checkpoint()跟cache()的原理

1. checkpoint() 方法和 cache() 方法都是 Spark中的缓存机制，用于提高计算效率的操作，都可以在迭代计算或长时间计算中使用，以减少计算时间和提高数据可靠性，但是它们的实现方式不同。
2. cache() 是将数据缓存在内存中，优势在于速度快，但缺点在于内存有限，数据可能会被清除；
3. 而 checkpoint()则是将数据写入磁盘中，优势在于数据可靠性高，但缺点在于速度慢，需要写入磁盘。
4. cache() 是一个转换操作，不会立即执行计算，只有在行动操作需要使用数据时才会触发计算。
5. checkpoint()会触发一次完整的计算，并将结果写入到磁盘中，因此它是一个行动操作。
6. 但是，checkpoint() 方法需要将数据写入磁盘或分布式文件系统中，这会导致额外的 I/O操作，影响性能。因此，为了避免频繁地进行 I/O 操作，通常将 checkpoint() 方法和 cache() 方法结合使用。
7. 具体来说，可以先使用 cache() 方法将数据缓存到内存中，然后再使用 checkpoint()方法将数据写入磁盘或分布式文件系统中。这样可以避免频繁地进行 I/O 操作，同时又能保证数据的可靠性和可恢复性。
8. 需要注意的是，使用 checkpoint()会将数据写入磁盘或分布式文件系统中，这会占用一定的存储空间。因此，需要根据实际情况来决定何时使用 checkpoint()方法，以避免浪费存储资源。

3、checkpoint()和cache()结合时，谁前谁后呢？

在 PyTorch 中，checkpoint() 和 cache() 都是模型优化中经常使用的函数，但它们的使用方式和作用不同。

checkpoint() 函数是一种优化方法，可以将模型的一部分计算推迟到后面执行，从而减少显存的占用，提高模型的训练速度。
而 cache() 函数是一种缓存方法，可以将模型的某些计算结果缓存下来，以便下次使用时可以直接调用，避免重复计算，提高模型的训练速度。

这两个函数的使用顺序取决于具体的场景。
如果你希望先缓存模型的某些计算结果，再对模型进行优化，那么就应该先使用 cache() 函数，再使用 checkpoint() 函数。
如果你希望先对模型进行优化保存，再将优化后的结果缓存下来，那么就应该先使用 checkpoint() 函数，再使用 cache() 函数。