java - rdd.checkpoint 在 spark 作业中被跳过

Question

您好，我正在尝试运行一个经常因 StackoverflowError 而失败的长 sparkjob。该作业读取一个 parquetfile 并在 foreach 循环中创建一个 rdd。在做了一些研究之后，我认为为每个 rdd 创建一个检查点将帮助我解决我的内存问题。 (我尝试了不同的内存、开销内存、并行性、重新分区，并找到了最适合该作业的设置，但有时它仍然会失败，具体取决于我们集群上的负载。)

现在谈谈我真正的问题。我正在尝试创建检查点，方法是首先读取创建 RDD 的 Parquet ，然后缓存它，运行检查点函数，然后首先调用操作以使检查点发生。在我指定的路径中没有创建任何检查点，它表示跳过该阶段的 YARN UI。谁能帮我理解这个问题:)

  ctx.getSparkContext().setCheckpointDir("/tmp/checkpoints");
    public static void writeHhidToCouchbase(DataFrameContext ctx, List<String> filePathsStrings)  {
    filePathsStrings
        .forEach(filePath -> {
          JavaPairRDD<String, String> rdd =
              UidHhidPerDay.getParquetFromPath(ctx, filePath);
          rdd.cache();
          rdd.checkpoint();
          rdd.first();
          rdd.foreachPartition(p -> {
            CrumbsClient client = getClient();
            p.forEachRemaining(uids -> {
              Crumbs crumbs = client.getAsync(uids._1)
                  .timeout(10, TimeUnit.SECONDS)
                  .toBlocking()
                  .first();
              String hHid = uids._2;
              if (hHid != null) {
                crumbs.getOrCreateSingletonCrumb(HouseholdCrumb.class).setHouseholdId(hHid);
                client.putSync(crumbs);
              }
            });
            client.shutdown();
          });
        });
}

检查点在第一次迭代中创建一次，但不会再次创建。韩国

Answer 1

我的错误是实际上创建了分区。我上面提到的“第一个”分区是一个包含分区的目录。由于像 8f987639-d5c8-46b8-a1e0-37081f9f8e00 这样的目录名称，我感到困惑。然而，查看来自@ImDarrenG 的血统评论给了我更多的见解。我从我缓存和检查点的第一个 RDD 创建了一个新的重新分区的 RDD。这使应用程序更加稳定，没有出现故障。

JavaPairRDD<String, String> rdd =
          UidHhidPerDay.getParquetFromPath(ctx, filePath);
      rdd.cache();
      rdd.checkpoint();
      rdd.first();
      JavaPairRDD<String, String> rddToCompute = rdd.repartition(72);
      rddToCompute.foreachPartition...