3

我正在将我的谷歌数据流 java 1.9 迁移到梁 2.0，我正在尝试使用 BigtableIO.Write

    ....
.apply("", BigtableIO.write()
                .withBigtableOptions(bigtableOptions)
                .withTableId("twoSecondVitals"));

在 BigtableIO 之前的 ParDo 中，我正在努力使 Iterable 成为可能。

          try{
        Mutation mutation = Mutation.parseFrom(new ObjectMapper().writeValueAsBytes(v));
        Mutation mu[] = {mutation};
        Iterable<Mutation> imu = Arrays.asList(mu);
        log.severe("imu");
        c.output(KV.of(ByteString.copyFromUtf8(rowKey+"_"+v.getEpoch()), imu));
      }catch (Exception e){
        log.severe(rowKey+"_"+v.getEpoch()+" error:"+e.getMessage());
      }

上面的代码抛出以下异常 InvalidProtocolBufferException: Protocol message end-group tag did not match expected tag

v 是一个对象列表（Vitals.class）。hbase api 使用 Put 方法来创建突变。如何创建一个可与 BigtableIO 接收器一起使用的 BigTable 突变？

4

1 回答 1

4

通过查看 sdk 的测试，我能够找到我的答案。 

            Iterable<Mutation> mutations =
                ImmutableList.of(Mutation.newBuilder()
                .setSetCell(
                        Mutation.SetCell.newBuilder()
                        .setValue(ByteString.copyFrom(new ObjectMapper().writeValueAsBytes(v)))
                        .setFamilyName("vitals")
                ).build());
于 2017-06-26T20:40:38.127 回答