scoobi 连接 DLists 问题

Question

我有以下结构的 scala scoobi 代码（我现在在本地执行）：

def run() = {
  val myvalue1 = processSource.source1("mypath/inputData/", references) // has several subfolders with input files that should be processed and some lines in these files are to be rejected due to some reasons and written to a log
  val (partitionedVal1, partitionedVal2) =  myvalue1.partition {
    case Left(x) => true
    case _ => false}
  val output = processData.process(partitionedVal1) // some processing
  val getRidOfRight = partitionedVal2.map(x => x.right.get)
  persist(getRidOfRight.toTextFile("output/log", overwrite = true), output.toTextFile("output/processed_data", overwrite = true))

}

并且 processSource 是一个对象：

object processSource {
def source1(a:DList[Either[Class1, String]] , b:DList[Either[Class2, String]] ) = {
val (a1, a2) = a.partition {
        case Left(x) => true
        case _ => false}
val (b1, b2) = b.partition {
        case Left(x) => true
        case _ => false}
val joinedAB1:DList[Either[Class1, String]] = a1.by(_.myKey).joinLeft(b1.by(_.myOtherKey)).map {
case (key, (val1, Some(val2))) => Left(val1.withCommon(val2))
case (key,(val1, None)) => Left(val1)}

val AB2:DList[Either[Class1, String]] = b2.map{ x => Right(x.right.get)}

AB1 ++ AB2 ++ b1
   } 
}

现在，我的问题是 getRidOfRight return 只是它应该 return 的输出的一部分。为什么？

Answer 1

github.com/NICTA/scoobi/issues/328 它是 scoobi 中的一个错误 - 解决方案是使用 scoobi 0.9.1 和 "collect" 而不是 "partition"（"partition" 未修复）

scoobi 连接 DLists 问题

scoobi concatenate DLists issue

scala

persist

partition

either