如何将 Scalding TypedPipe 转换为 Iterator答案

【问题标题】：how to convert Scalding TypedPipe to Iterator如何将 Scalding TypedPipe 转换为 Iterator
【发布时间】：2015-12-07 01:11:09
【问题描述】：

在我的 Scalding hadoop 工作中，我在管道上有一些分组逻辑，然后我需要处理每个组：

val georecs : TypedPipe[GeoRecord] = getRecords

georecs.map( r => (getRegion(r),r) )
  .groupBy(_._1)
  .mapValueStream( xs => clusterRecords(xs) )
  .values
  .write(out)

在 clusterRecords 中，我需要将传入的迭代器转换为 TypedPipe，以便 1) 对其进行采样并 2) 获取叉积：

//turn the iterator to a pipe so we can sample it    
    val sample = TypedPipe.from( xs.map( x => Centroid(x._2.coreActivity)).toIterable)
    .sample(0.11)
    .distinct

//turn the iterator to a pipe so we can take its cross product
val records : TypedPipe[GeoRecord] = TypedPipe.from(xs.map(_._2).toIterable)

records
  .cross(sample)   //cartesian product of records and centroids
  .groupBy( _._2)  // group By the user record so we get a list of pairs (user, centroid)
  .minBy( x => score( x._1.coreActivity, x._2.core) ) //find the centroid with the lowest score for each Record
  .values
  .groupBy( x => x._2 )   //now groupBy centroid to get the clusters
  .values

问题是 mapValueStream 期望映射函数返回一个迭代器，但我拥有的是一个 TypedPipe。我知道如何将迭代器变成管道，但反过来却不行。我是否需要执行它，将其写入磁盘，然后再将其读回？

如果是这样，最好的方法是什么？

【问题讨论】：

标签： scala hadoop iterator pipe scalding

【解决方案1】：

看起来您可以通过运行将管道转换为迭代器。可以这样完成：

val georecs : TypedPipe[GeoRecord] = getRecords

val i : Iterator[GeoRecord] = georecs
  .toIterableExecution
  .waitFor(this.scaldingConfig,this.mode)
  .get
  .toIterator

（类型检查，但尚未测试）

【讨论】：