使用 scalding 读取多个文件并输出一个 SINGLE 文件答案

【问题标题】：Read multiple files using scalding and output a SINGLE file使用 scalding 读取多个文件并输出一个 SINGLE 文件
【发布时间】：2017-05-02 04:06:59
【问题描述】：

这些天我遇到了一个问题，我正在尝试使用 scalding 从多个文件中读取并使用单个文件创建输出。我的代码是这样的：

def getFilesSource (paths: Seq[String]) = {
    new MultipleTextLineFiles(paths: _*) {
      override protected def createHdfsReadTap(hdfsMode: Hdfs): Tap[JobConf, _, _] =  {
        val taps = goodHdfsPaths(hdfsMode).toList.map {
          path => CastHfsTap (new Hfs (hdfsScheme, path, sinkMode))
        }

        taps.size match {
          case 0 => {
            CastHfsTap (new Hfs(hdfsScheme, hdfsPaths.head, sinkMode))
          }
          case 1 => taps.head
          case _ => new ScaldingMultiSourceTap(taps)
        }
      }
    }
  }

但是当我运行此代码时，它会将我的输出拆分为许多文件，但里面的数据非常少：只有几 K。相反，我希望能够将所有输出文件聚合到一个文件中。

我的烫伤代码是：

val source = getFilesSource(mapped) // where mapped is a Sequence of valid HDFS paths (Seq [String])

TypedPipe.from(source).map(a => Try{
  val json = JSON.parseObject(a)
  (json.getInteger("prop1"), json.getInteger("prop2"), json.getBoolean("prop3"))
}.toOption).filter(a => a.nonEmpty)
  .map(a => a.get)
  .filter(a => !a._3)
  .map (that => MyScaldingType (that._1, that._2))
  .write(MyScaldingType.typedSink(typedArgs))

我想我必须重写 ScaldingMultiSourceTap 类型的“sourceConfInit”方法，但我不知道在里面写什么......

【问题讨论】：

标签： scala hadoop hdfs hadoop2 scalding

【解决方案1】：

您可以使用 groupAll 将所有地图输出（该作业是仅地图作业）发送到单个减速器，考虑到数据很小，然后进行写入。输出将被写入单个文件。

.
.
.
.filter(a => !a._3)
.map (that => MyScaldingType (that._1, that._2))
.groupAll
.write(MyScaldingType.typedSink(typedArgs))

【讨论】：

嗨@karthikcru，感谢您的回答，听起来很有希望。我今天早上会试试这个。我希望由于我在映射阶段执行过滤器（使用语句： .filter(a => !a._3) ），对于我的业务案例，会有很多数据不会通过该过滤器标准。剩余的东西将被发送到单个减速器。