title: how to convert Scalding TypedPipe to Iterator
body: <p>In my Scalding hadoop job, I've got some grouping logic on a pipe, and then I need to process each group:</p>
  
  <pre><code>val georecs : TypedPipe[GeoRecord] = getRecords
  
  georecs.map( r =&gt; (getRegion(r),r) )
    .groupBy(_._1)
    .mapValueStream( xs =&gt; clusterRecords(xs) )
    .values
    .write(out)
  </code></pre>
  
  <p>Inside clusterRecords I need to convert the iterator that's passed in into a TypedPipe so that I can 1) sample it and 2) take the cross product:</p>
  
  <pre><code>//turn the iterator to a pipe so we can sample it    
      val sample = TypedPipe.from( xs.map( x =&gt; 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 =&gt; score( x._1.coreActivity, x._2.core) ) //find the centroid with the lowest score for each Record
    .values
    .groupBy( x =&gt; x._2 )   //now groupBy centroid to get the clusters
    .values
  </code></pre>
  
  <p>The problem is that mapValueStream expects the mapping function to return an iterator, but what I have is a TypedPipe. I know how to turn an iterator into a pipe, but not the other way around. Do I need to execute it, write it to disk, and then read it back in? </p>
  
  <p>And if so, what's the best way to accomplish that?</p>