java - 如何在 Scalding 中存储输出

Question

我正在尝试将一个管道输出到不同的目录中，这样每个目录的输出将根据一些 ID 进行分桶。所以在普通的 map reduce 代码中，我会使用 MultipleOutputs 类，我会在 reducer 中做类似的事情。

protected void reduce(final SomeKey key,
      final Iterable<SomeValue> values,
      final Context context) {

   ...
   for (SomeValue value: values) {
     String bucketId = computeBucketIdFrom(...);
     multipleOutputs.write(key, value, folderName + "/" + bucketId);
   ...

所以我想在烫伤时可以这样做

...
  val somePipe = Csv(in, separator = "\t",
        fields = someSchema,
        skipHeader = true)
    .read

  for (i <- 1 until numberOfBuckets) {
    somePipe
    .filter('someId) {id: String => (id.hashCode % numberOfBuckets) == i}
    .write(Csv(out + "/bucket" + i ,
      writeHeader = true,
      separator = "\t"))
  }

但我觉得你最终会多次重写同一个管道，这会影响整体性能。

还有其他选择吗？

谢谢

Answer 1

是的，当然有更好的方法使用 TemplatedTsv .

所以你上面的代码可以这样写，

val somePipe = Tsv(in, fields = someSchema, skipHeader = true)
    .read
    .write(TemplatedTsv(out, "%s", 'some_id, writeHeader = true))

这会将来自 'some_id 的所有记录放入 out/some_ids 文件夹下的单独文件夹中。

但是，您也可以创建整数桶。只需更改最后几行，

.map('some_id -> 'bucket) { id: String => id.hashCode % numberOfBuckets }    
.write(TemplatedTsv(out, "%02d", 'bucket, writeHeader = true, fields = ('all except 'bucket)))

这将创建两位数字的文件夹，如 out/dd/。您还可以查看 templatedTsv api here.

使用 templatedTsv 可能会出现小问题，即 reducer 会生成大量小文件，这可能对使用您的结果的下一个工作不利。因此，最好在写入磁盘之前对模板字段进行排序。我写了一篇关于它的博客 here.