Pub/Sub Topic的Avro Schema映射BigQuery表时类型不兼容问题

我有一个附加了以下Avro Schema的Pub/Sub Topic，希望通过设置use_topic_schema为true的推送至BigQuery（push-to-bq）类型订阅，将Topic数据映射到BigQuery表Schema。

关联的Avro Schema #

[
  {
    "name" : "columnA",
    "type" : [ {
      "type" : "bytes",
      "logicalType" : "decimal",
      "precision" : 18,
      "scale" : 4
    }, "null" ]
  }, {
    "name" : "columnB",
    "type" : [ {
      "type" : "bytes",
      "logicalType" : "decimal",
      "precision" : 19,
      "scale" : 4
    }, "null" ]
  }, {
    "name" : "columnC",
    "type" : [ {
      "type" : "bytes",
      "logicalType" : "decimal",
      "precision" : 38,
      "scale" : 6
    }, "null" ]
  }
]

尝试的BigQuery表Schema #

[
  {
    "mode": "NULLABLE",
    "name": "columnA",
    "type": "FLOAT"
  },
  {
    "mode": "NULLABLE",
    "name": "columnB",
    "type": "NUMERIC"
  },
  {
    "mode": "NULLABLE",
    "name": "columnC",
    "type": "FLOAT"
  }
]

报错信息 #

│ Error: Error creating Subscription: googleapi: Error 400: Incompatible schema type for field NetSalesLocalCurrency: BYTES vs. FLOAT
│ Details:
│ [
│   {
│     "@type": "type.googleapis.com/google.rpc.ErrorInfo",
│     "domain": "pubsub.googleapis.com",
│     "metadata": {
│       "actual_value": "FLOAT",
│       "expected_value": "BYTES",
│       "field_name": "ColumnC",
│       "reason": "INCOMPATIBLE_TYPE"
│     },
│     "reason": "INCOMPATIBLE_SCHEMA"
│   }
│ ]

报错原因是Pub/Sub推送到BigQuery时，会匹配Avro的原始物理类型而非逻辑类型。你的Avro字段物理类型是bytes，但指定的BigQuery类型是FLOAT/NUMERIC，类型不兼容。

根据decimal字段的精度，对应BigQuery的正确类型如下：

• columnA（精度18）：用NUMERIC（BigQuery NUMERIC支持最高28位精度，满足需求）
• columnB（精度19）：用BIGNUMERIC（支持最高38位精度）
• columnC（精度38）：用BIGNUMERIC（匹配最大精度）

调整后的BigQuery Schema：

[
  {
    "mode": "NULLABLE",
    "name": "columnA",
    "type": "NUMERIC"
  },
  {
    "mode": "NULLABLE",
    "name": "columnB",
    "type": "BIGNUMERIC"
  },
  {
    "mode": "NULLABLE",
    "name": "columnC",
    "type": "BIGNUMERIC"
  }
]

开启use_topic_schema后，Pub/Sub会自动基于Avro的decimal逻辑类型，把bytes转换为对应的BigQuery数值类型，无需手动处理字节数据。

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