BigQuery推送至Pub/Sub无消息排查：TableRow转JSON方法咨询

你的代码没报错但没发消息，核心问题是错误地将整个TableRow对象序列化为JSON，而不是直接提取requestPayload列里的原始JSON字符串。

原代码的问题 #

AsJsons.of(TableRow.class)会把TableRow转换成包含字段名的JSON结构，比如BigQuery里的requestPayload值是{"user":"test"}，转换后会变成：

{"requestPayload":"{\"user\":\"test\"}"}

这不是你需要的原始JSON内容，而且如果处理逻辑没适配，可能导致数据没被正确输出（比如空值过滤缺失）。

正确的实现方式 #

直接提取TableRow中requestPayload字段的字符串值，不需要额外的JSON转换（因为它本身就是合法的JSON字符串），修改代码如下：

p.apply("ReadSourceBQ",
    BigQueryIO
        .readTableRows()
        .fromQuery("SELECT requestPayload FROM eventreplay")
        .usingStandardSql()
        .withTemplateCompatibility()
)
.apply("Extract Request Payload", ParDo.of(new DoFn<TableRow, String>() {
    @ProcessElement
    public void processElement(ProcessContext c) {
        // 提取requestPayload字段的原始字符串
        String rawPayload = (String) c.element().get("requestPayload");
        // 跳过空值或无效数据，避免发送空消息
        if (rawPayload != null && !rawPayload.trim().isEmpty()) {
            c.output(rawPayload);
        }
    }
}))
.apply("WriteToPubSub", PubsubIO.writeStrings().to("topicname"));

额外检查项 #

• 确认BigQuery查询SELECT requestPayload FROM eventreplay确实返回数据，可以在BigQuery控制台手动执行验证。
• 检查Dataflow服务账号是否拥有Pub/Sub主题的pubsub.topics.publish权限。
• 查看Dataflow作业的日志，确认数据是否被正确提取和输出（比如有没有跳过空值的日志）。

内容的提问来源于stack exchange，提问作者user2495833