Apache Beam TypeScript SDK指定BigQuery Schema及报错解决

问题原因 #

你遇到的The input doesn't has a schema错误，本质是从PubSub读取的原始数据（字符串/字节）没有明确的结构，Beam无法自动映射到目标BigQuery表的Schema。虽然createDisposition: "Never"表示不创建新表，但Beam仍需要输入数据的结构与已存在的BigQuery表Schema匹配，而TypeScript SDK的writeToBigQuery不需要显式传入Schema参数，而是依赖输入的PCollection元素为结构化JS对象来完成映射。

解决方案 #

核心步骤是将PubSub的原始消息解析为与目标BigQuery表Schema完全匹配的结构化对象，具体实现如下：

1. 明确目标BigQuery表的Schema

假设你的BigQuery表有如下Schema：

2. 修改管道代码，添加消息解析逻辑

将PubSub的原始字符串消息解析为对应结构的JS对象，确保字段名、类型与BigQuery表一致：

import * as beam from "apache-beam";
import * as bigqueryio from "apache-beam/io/bigqueryio";

export function createPipeline() {
  return async (root: beam.Root) => {
    // 读取PubSub原始消息（默认返回字符串类型）
    const rawPubSubMessages = await root.applyAsync(
      beam.readFromPubSub({
        topic: "projects/PROJECT_NAME/topics/TOPIC_NAME",
      })
    );

    // 解析消息为结构化对象，匹配BigQuery表Schema
    const structuredData = rawPubSubMessages.apply(
      beam.map((message: string) => {
        try {
          // 假设PubSub消息是JSON格式，解析为对象
          const rawData = JSON.parse(message);
          // 转换字段类型以匹配BigQuery表
          return {
            id: Number(rawData.id), // 转为数字对应INT64
            content: rawData.content,
            event_time: new Date(rawData.event_time).toISOString() // 转为ISO格式对应TIMESTAMP
          };
        } catch (error) {
          // 处理解析失败的消息，可选择丢弃或标记
          console.error("消息解析失败:", error, "原始消息:", message);
          return null;
        }
      })
    );

    // 过滤解析失败的无效消息
    const validData = structuredData.apply(
      beam.filter((item) => item !== null)
    );

    // 写入BigQuery
    await validData.applyAsync(
      bigqueryio.writeToBigQuery("PROJECT_NAME:DATASET.TABLE_NAME", {
        createDisposition: "Never",
        writeDisposition: "WriteAppend", // 根据需求选择追加/覆盖
      })
    );
  };
}

额外注意事项 #

• 若PubSub消息是二进制格式，需先转换为UTF-8字符串：Buffer.from(message).toString('utf8')
• 确保结构化对象的字段名、类型与BigQuery表完全一致（比如BigQuery的TIMESTAMP需对应ISO格式字符串或Date对象）
• 若目标表Schema有嵌套字段，需在解析时生成对应的嵌套JS对象

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