Terraform配置Pub/Sub死信主题写入BigQuery及消息重放问题

1. Terraform中字段配置的块位置 #

是的，需要在google_pubsub_subscription资源的bigquery_config块中开启元数据写入，同时必须提前用Terraform创建符合字段要求的BigQuery表，确保表结构兼容Pub/Sub推送的元数据。

2. Terraform配置BQ表与Pub/Sub订阅示例 #

步骤1：创建BigQuery数据集与表 #

先定义BQ表的Schema，包含要求的google_bigquery_table、message_id、publish_time、data、attributes字段：

resource "google_bigquery_dataset" "dl_dataset" {
  dataset_id = "dead_letter_dataset"
  project    = var.project_id
  location   = "us-central1"
}

resource "google_bigquery_table" "dl_table" {
  dataset_id = google_bigquery_dataset.dl_dataset.dataset_id
  table_id   = "dead_letter_messages"
  project    = var.project_id

  schema = <<EOF
[
  {
    "name": "google_bigquery_table",
    "type": "STRING",
    "mode": "NULLABLE"
  },
  {
    "name": "message_id",
    "type": "STRING",
    "mode": "REQUIRED"
  },
  {
    "name": "publish_time",
    "type": "TIMESTAMP",
    "mode": "REQUIRED"
  },
  {
    "name": "data",
    "type": "STRING",
    "mode": "NULLABLE"
  },
  {
    "name": "attributes",
    "type": "RECORD",
    "mode": "NULLABLE",
    "fields": [
      {
        "name": "key",
        "type": "STRING",
        "mode": "NULLABLE"
      },
      {
        "name": "value",
        "type": "STRING",
        "mode": "NULLABLE"
      }
    ]
  }
]
EOF
}

步骤2：创建死信主题与订阅 #

创建dl_topic死信主题，再创建dl_subs订阅关联上述BQ表，开启元数据写入：

resource "google_pubsub_topic" "dl_topic" {
  name    = "dl_topic"
  project = var.project_id
}

resource "google_pubsub_subscription" "dl_subs" {
  name               = "dl_subs"
  topic              = google_pubsub_topic.dl_topic.name
  project            = var.project_id

  bigquery_config {
    table                  = google_bigquery_table.dl_table.id
    write_metadata         = true
    use_topic_schema       = false
    drop_unknown_fields    = false
  }
}

3. 从BigQuery重放消息至Pub/Sub #

方法1：gcloud命令行批量重放 #

• 从BQ导出数据到Cloud Storage（JSON格式）：

  bq extract --destination_format NEWLINE_DELIMITED_JSON dead_letter_dataset.dead_letter_messages gs://your-bucket/dl-messages.json
  

• 编写脚本读取GCS文件并发送消息（需安装jq）：

  gsutil cat gs://your-bucket/dl-messages.json | while read line; do
    DATA=$(echo $line | jq -r '.data')
    ATTRIBUTES=$(echo $line | jq -r '.attributes | to_entries | map("\(.key)=\(.value)") | join(",")')
    gcloud pubsub topics publish your-target-topic --message="$DATA" --attributes="$ATTRIBUTES"
  done
  

方法2：Dataflow自动化重放 #

创建Dataflow作业，读取BQ表中的消息数据，转换为Pub/Sub兼容格式后批量发送到目标主题，适合大规模消息重放场景。

方法3：自定义代码重放 #

用Python/Java等语言编写程序，通过BQ客户端读取数据，再通过Pub/Sub客户端发布消息。示例Python片段：

from google.cloud import bigquery
from google.cloud import pubsub_v1

bq_client = bigquery.Client()
pubsub_publisher = pubsub_v1.PublisherClient()
topic_path = pubsub_publisher.topic_path("your-project", "your-target-topic")

query = "SELECT data, attributes FROM `dead_letter_dataset.dead_letter_messages`"
results = bq_client.query(query)

for row in results:
    data = row["data"].encode("utf-8")
    attributes = {item["key"]: item["value"] for item in row["attributes"]}
    pubsub_publisher.publish(topic_path, data, **attributes).result()

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