使用 apache beam dataflow java 流式传输 pubsub -bigtable

Question

尝试将 pubsub json 消息更新为 bigtable。我正在从本地计算机运行代码。正在创建数据流作业。但我没有看到 bigtable 实例中更新了任何数据，而且它也不会在控制台或数据流作业中引发任何错误。我还尝试了硬编码值并尝试在 bigtable 中更新，但仍然不起作用。请任何人在这个问题上建议或指导我

 try{
    PipelineOptions options = PipelineOptionsFactory.fromArgs(projectArgs).create();
    options.setRunner(DataflowRunner.class);
    System.out.println("tempfile-" + options.getTempLocation());
    Pipeline p = Pipeline.create(options);
    System.out.println("options" + options.getTempLocation());
    p.apply("Read PubSub Messages", PubsubIO.readStrings().fromTopic(PUBSUB_SUBSCRIPTION))
            .apply(Window.into(FixedWindows.of(Duration.standardMinutes(1))))
            .apply(ParDo.of(new RowGenerator())).apply(CloudBigtableIO.writeToTable(bigtableConfig));

    p.run();
    }catch (Exception e) {
        // TODO: handle exception
        System.out.println(e);
    }
}

@ProcessElement public void processElement(ProcessContext context) {

try {
System.out.println("In for RowGenerator");
String decodedMessageAsJsonString = context.element();
System.out.println("decodedMessageAsJsonString"+decodedMessageAsJsonString);

String rowKey = String.valueOf(
        LocalDateTime.ofInstant(Instant.now(), ZoneId.of("UTC"))
                .toEpochSecond(ZoneOffset.UTC));
System.out.println("rowKey"+rowKey);

Put put = new Put(rowKey.getBytes());
put.addColumn("VALUE".getBytes(), "VALUE".getBytes(), decodedMessageAsJsonString.getBytes());
     // put.addColumn(Bytes.toBytes("IBS"), Bytes.toBytes("name"),Bytes.toBytes("ram"));

context.output(put);
}catch (Throwable e) {
    // TODO: handle exception
    System.out.println(e);
}

}在此处输入图像描述

Answer 1

我没有看到模板的 Bigtable 方面有任何问题。只需确保目标表上存在列族（我假设为“VALUE”）。

您确定您正在阅读正确的 PubSub 订阅并且有消息发送到 PubSub。如果一切正确，则似乎存在PubSub 配置中存在一些问题，也许可以在问题上添加 PubSub 标签，并且来自 pubsub 社区的人可以提供帮助。