2

我对 Cascading/Scalding 非常陌生，并且无法弄清楚从 HBase 读取数据的热度。

我在 HBase 中有一张表，其中存储了扑克游戏的手牌历史记录（以非常简单的方式：）id -> hand, serialized with ProtoBuf。下面的工作应该遍历整个历史，并建立一个所有玩家的字典：

class DictionaryBuilder(args: Args) extends Job(args) {

  val input = new HBaseSource("hand", "localhost", 'hand, Array("d"), Array("blob"))
  val output = TextLine("tutorial/data/output0.txt")

  input
    .read
    .flatMap('hand -> 'player) {
    handBytes: Array[Byte] =>
       HandHistory.parseFrom(handBytes).getPlayerList.map(_.getName)
    }
    .write(output)

}

但是，当我运行上面的作业时，会抛出错误

Caused by: com.google.protobuf.InvalidProtocolBufferException: Protocol message end-group tag did not match expected tag.
    at com.google.protobuf.InvalidProtocolBufferException.invalidEndTag(InvalidProtocolBufferException.java:73)
    at com.google.protobuf.CodedInputStream.checkLastTagWas(CodedInputStream.java:106)
    at com.google.protobuf.AbstractMessageLite$Builder.mergeFrom(AbstractMessageLite.java:163)
    at com.google.protobuf.AbstractMessage$Builder.mergeFrom(AbstractMessage.java:661)

，这意味着得到的数据flatMap不是我可以直接使用的字节数组。

我错过了什么？

4

1 回答 1

1

看看这个项目https://github.com/ParallelAI/SpyGlass，它为 Scalding 提供了一个 HBase tap。顺便说一句，如果你想避免处理Array[Byte]，使用SpyGlass，你可以使用fromBytesWritable方法来转换你需要处理的字段。

于 2014-04-20T22:26:56.027 回答