Spark：RDD在每次迭代中缺少条目

Question

我正在尝试实施一种自学习方法来训练分类器。我正在使用火花 1.6.0。问题是当我将一个 RDD 映射到另一个时，我得到了错误的计数。相同的代码适用于小型数据集，但在较大的数据集上它只是疯了。

println("INITIAL TRAINING SET SIZE : " + trainingSetInitial.count())
for(counter <- 1 to 10){
  println("-------------------  This is the_" + counter + " run -----------------")
  println("TESTING SET SIZE : "  + testing.count())

  val lowProbabilitiesSet = testing.flatMap { item =>
    if (model.predictProbabilities(item._2)(0) <= 0.75 && model.predictProbabilities(item._2)(1) <= 0.75) {
      List(item._1)
    } else {
      None
    }}.cache()
  val highProbabilitiesSet = testing.flatMap { item =>
    if (model.predictProbabilities(item._2)(0) > 0.75 || model.predictProbabilities(item._2)(1) > 0.75 ) {
      List(item._1 +","+ model.predict(item._2).toDouble )
    } else {
      None
    }}.cache()
  println("LOW PROBAB SET : "  + lowProbabilitiesSet.count())
  println("HIGH PROBAB SET : "  + highProbabilitiesSet.count())

  trainingSetInitial = trainingSetInitial.union(highProbabilitiesSet.map(x => LabeledPoint(List(x)(0).split(",")(8).toString.toDouble, htf.transform(List(x)(0).toString.split(",")(7).split(" ") ))))
  model = NaiveBayes.train(trainingSetInitial, lambda = 1.0)
  println("NEW TRAINING SET : "  + trainingSetInitial.count())

  previousCount = lowProbabilitiesSet.count()
  testing = lowProbabilitiesSet.map { line =>
    val parts = line.split(',')
    val text = parts(7).split(' ')
    (line, htf.transform(text))
  }
  testing.checkpoint()
}

这是来自正确输出的日志：

初始训练集大小：238.182

------------------- 这是_1 运行------

测试集大小：3.158.722

低概率集：22.996

高概率集：3.135.726

新训练套装：3373908

------------------- 这是_2 运行------

测试集尺寸：22996

低概率集：566

高概率集：22430

新训练套装：3396338

这是问题开始的时候（大数据集输入）：

初始训练集尺寸： 31.990.660

------------------- 这是_1 运行------

测试集尺寸：423.173.780

低概率集：62.615.460

高概率集：360.558.320

新训练集：395265857

------------------- 这是_2 运行------

测试集尺寸：52673986

低概率集：51460875

高概率集：1213111

新训练套装：401950263

第一次迭代的“LOW PROBAB SET”应该是第二次迭代的“TESTING SET”。不知何故，1000 万个条目消失了。第一次迭代的“新训练集”也应该是“初始训练”和“高概率集”的串联。再次，数字不匹配。

代码运行时我没有收到任何错误。我尝试缓存每个集合并在每次迭代结束时取消持久化（仅限 HIGH 和 LOW 集合），但结果相同。我也试图检查集合，没有工作。为什么会这样？

编辑

只是为了测试，我没有在循环内创建新模型只是为了看看会发生什么：

for(counter <- 1 to 5){
  println("-------------------  This is the_" + counter + " run !!! -----------------")
  var updated_trainCnt = temp_train.count();
  var updated_testCnt = test_set.count();
  println("Updated Train SET SIZE: "  + updated_trainCnt)
  println("Updated Testing SET SIZE: "  + updated_testCnt)

  val highProbabilitiesSet = test_set.filter { item =>
    val output = model.predictProbabilities(item._2)
    output(0) > 0.75 || output(1) > 0.75
  }.map(item => (item._1 + "," + model.predict(item._2), item._2 )).cache()

  test_set = test_set.filter { item =>
    val output = model.predictProbabilities(item._2)
    output(0) <= 0.75 && output(1) <= 0.75
  }.map(item => (item._1, item._2)).cache()
  var hiCnt = highProbabilitiesSet.count()
  var lowCnt = test_set.count()
  println("HIGH PROBAB SET : "  + hiCnt)
  println("LOW PROBAB SET  : "  +  lowCnt)
  var diff = updated_testCnt - hiCnt - lowCnt
  if (diff!=0) println("ERROR: Test set not correctly split into high low" + diff)
  temp_train= temp_train.union(highProbabilitiesSet.map(x => LabeledPoint(x._1.toString.split(",")(8).toDouble, x._2 ))).cache()
  println("NEW TRAINING SET: "  + temp_train.count())
//      model = NaiveBayes.train(temp_train, lambda = 1.0, modelType = "multinomial")
  println("HIGH PROBAB SET : "  + highProbabilitiesSet.count())
  println("LOW PROBAB SET  : "  + test_set.count())
  println("NEW TRAINING SET: "  + temp_train.count())
}

从原始模型生成的数字是可以的，即使 RDD 的联合执行没有问题。但是最大的问题仍然存在，分类模型如何在每个循环结束时（或其他 RDD）都没有修改训练集（lowProbabilititesSet）？

控制台日志和火花日志不显示任何错误或刽子手粉碎。分类训练过程如何破坏我的数据？

Answer 1

尽管我仍然没有弄清楚为什么会发生这种情况，但我还是将 RDD 刷新到 HDFS 并制作了一个 bash 脚本，该脚本迭代地运行该类并每次都从 HDFS 读取数据。当我在循环内训练分类器时，我发现问题出现了。

Answer 2

我没有立即发现问题。请将代码最小化为实际问题。首先我建议将flatMap 操作重写为filter：

val highProbabilitiesSet = testing.flatMap { item =>
  if (model.predictProbabilities(item._2)(0) > 0.75 || model.predictProbabilities(item._2)(1) > 0.75 ) {
      List(item._1 +","+ model.predict(item._2).toDouble )
  } else {
    None
  }
}.cache()

收件人：

val highProbabilitiesSet = testing.filter { item => 
  val output = model.predictProbabilities(item._2)
  output(0) > 0.75 || output(1) > 0.75
}.map(item => (item._1, model.predict(item._2).toDouble)).cache()