迭代后Spark数组是空的

当我迭代集合并在迭代遍布数组时将项添加到数组中似乎为空时，spark 1.6.0（我对spark和scala不太熟悉）。迭代后Spark数组是空的

var testing = unlabeled.map { line => 
    val parts = line.split(',') 
    val text = parts(7).split(' ') 
    (line, htf.transform(text)) 
} 

var lowPropQueue = new mutable.ArrayBuffer[(String, org.apache.spark.mllib.linalg.Vector)] 
var highPropQueue = new mutable.ArrayBuffer[(String, org.apache.spark.mllib.linalg.Vector)] 

for(counter <- 1 to 5){ 

    logger.info("this is the " + counter + " run -----------------") 
    for (i <- testing) { 
    val label = model.predict(i._2).toString 
    //  logger.info(i._1.split(",")(7)) 
    //  logger.info(label) 
    var probs = model.predictProbabilities(i._2) 
    logger.info("prob 0 : " + probs(0)) 
    logger.info("prob 1 : " + probs(1)) 
    logger.info("--------------------- ") 

    if (probs(0).toDouble <= 0.95 && probs(1).toDouble <= 0.95) { 
     lowPropQueue.+=(i) 
    } else { 
     highPropQueue.+=((i._1 + "," + label , i._2)) 
    } 

    logger.info("size of high array : " + highPropQueue.length) 
    logger.info("size of low array : " + lowPropQueue.length) 

    } 

    logger.info("passed: " + lowPropQueue.length) 
    logger.info("NOT passed: " + highPropQueue.length) 

    var xx= sc.parallelize(highPropQueue).collect() 
    var yy = sc.parallelize(lowPropQueue).collect() 

    logger.info("passed: " + xx.length) 
    logger.info("NOT passed: " + yy.length) 
... 
}

但是基于内环似乎元素添加到阵列中的日志，即：

16/10/11 11时22分31秒INFO SelfLearningMNB $：高数组大小： 500

16/10/11 11时22分31秒INFO SelfLearningMNB $：83

16/10/11 11时22分31秒INFO SelfLearningMNB $：低阵列的大小概率0：0.37094327822665185

16/10/11 11:22:31信息SelfLearningMNB $：概率1：0.6290567217733481

16/10/11 11:22:31信息SelfLearningMNB $：------------ ---------

16/10/11 11时22分31秒INFO SelfLearningMNB $：500

16/10/11 11时22分31秒INFO SelfLearningMNB $：高阵列的大小：84

16/10/11 11时22分31秒INFO SelfLearningMNB $：低阵列的大小概率0：0.16872929936216619

16/10/11 11时22分31秒INFO SelfLearningMNB $：概率1：0.8312707006378338

但是，当内循环结束我得到这个：

16/10/11 11时43分五十三秒INFO SelfLearningMNB $：通过：0

16/10/11 11:43:53信息SelfLearningMNB $：未通过：0

这是怎么回事？

编辑

你怎么会从执行者获取数据或保存从执行者到HDFS的数据，使他们能够从后面主节点读取？

来源

2016-10-11 bill

TL; DR这不能在Spark中工作。

这是怎么回事？

每个执行都有自己的lowPropQueue，highPropQueue副本。
迭代期间本地副本是否被修改
后迭代的本地副本将被丢弃

FYI天真追加到ArrayBuffer不是线程安全的。

来源

2016-10-11 10:09:55

我虽然这一点。但是如何将来自执行者的数据存储到“全局”数组呢？ – bill

您可以尝试累加器，但您需要同步访问权限并查看您的代码，但它不会扩展。 – 2016-10-11 10:21:07

我做了一些搜索，这种方法绝对不适合Spark。我不得不映射一切，但它的工作。 – bill

迭代后Spark数组是空的

回答

相关问题