java如何在烫伤中存储输出

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_id文件夹下的单独文件夹中

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

.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可能会有一个小问题，即还原程序可能会生成大量小文件，这可能会对使用结果的下一个作业不利。因此，最好在写入磁盘之前对模板字段进行排序。我写了一篇关于它的博客here.