如何向按摄入时间分区的BigQuery表上传指定_PARTITIONTIME的JSONL文件

我有事件数据正在流式传输到按摄入时间分区的BigQuery表中，摄入时间和事件时间戳几乎一致。但部分事件没有事件时间戳，所以没法按事件时间戳分区，只能用摄入时间折中。

现在有一份过去12个月的旧事件数据（约200GB，JSONL格式）需要上传到这个表，但遇到了问题：

尝试1：直接在JSON中包含_PARTITIONTIME字段 #

因为_PARTITIONTIME是BigQuery保留关键字，直接在schema中声明会报错，代码如下：

const metadata = {
    sourceFormat: 'NEWLINE_DELIMITED_JSON',
    schema: {
        fields: [
            {name: 'mail', type: 'JSON'},
            {name: 'delivery', type: 'JSON'},
            {name: '_PARTITIONTIME', type: 'TIMESTAMP'},
        ],
    },
    location: 'US',
};

尝试2：自定义partitionTimestamp字段 #

把JSON中的分区时间字段改名为partitionTimestamp，并在schema中配置时间分区，但报错说该字段不属于表结构，代码如下：

const metadata = {
    sourceFormat: 'NEWLINE_DELIMITED_JSON',
    schema: {
        fields: [
            {name: 'mail', type: 'JSON'},
            {name: 'delivery', type: 'JSON'},
            {name: 'partitionTimestamp', type: 'TIMESTAMP'},
        ],
        timePartitioning: {
            field: 'partitionTimestamp',
        },
    },
    location: 'US',
};

官方文档里没提到怎么向按摄入时间分区的表上传旧数据，求解决方法。

编辑1：目标表结构 #

schema: {
    fields: [
        {name: 'mail', type: 'JSON'},
        {name: 'delivery', type: 'JSON'},
    ] 
}

编辑2：示例JSON数据 #

其中delivery.timestamp和摄入时间近似相等，希望把这条记录写入对应的日期分区：

{
  "mail": {
    "timestamp": "2016-10-19T23:20:52.240Z",
    "source": "sender@example.com",
    "sourceArn": "arn:aws:ses:us-east-1:123456789012:identity/sender@example.com",
    "sendingAccountId": "123456789012",
    "messageId": "EXAMPLE7c191be45-e9aedb9a-02f9-4d12-a87d-dd0099a07f8a-000000",
    "destination": [
      "recipient@example.com"
    ],
    "headersTruncated": false,
    "headers": [
      {
        "name": "From",
        "value": "sender@example.com"
      },
      {
        "name": "To",
        "value": "recipient@example.com"
      },
      {
        "name": "Subject",
        "value": "Message sent from Amazon SES"
      },
      {
        "name": "MIME-Version",
        "value": "1.0"
      },
      {
        "name": "Content-Type",
        "value": "text/html; charset=UTF-8"
      },
      {
        "name": "Content-Transfer-Encoding",
        "value": "7bit"
      }
    ],
    "commonHeaders": {
      "from": [
        "sender@example.com"
      ],
      "to": [
        "recipient@example.com"
      ],
      "messageId": "EXAMPLE7c191be45-e9aedb9a-02f9-4d12-a87d-dd0099a07f8a-000000",
      "subject": "Message sent from Amazon SES"
    },
    "tags": {
      "ses:configuration-set": [
        "ConfigSet"
      ],
      "ses:source-ip": [
        "192.0.2.0"
      ],
      "ses:from-domain": [
        "example.com"
      ],
      "ses:caller-identity": [
        "ses_user"
      ],
      "ses:outgoing-ip": [
        "192.0.2.0"
      ],
      "myCustomTag1": [
        "myCustomTagValue1"
      ],
      "myCustomTag2": [
        "myCustomTagValue2"
      ]      
    }
  },
  "delivery": {
    "timestamp": "2016-10-19T23:21:04.133Z",
    "processingTimeMillis": 11893,
    "recipients": [
      "recipient@example.com"
    ],
    "smtpResponse": "250 2.6.0 Message received",
    "reportingMTA": "mta.example.com"
  }
}

针对你的场景，有三个实用方案，按推荐优先级排序：

方案1：用INSERT语句手动指定分区时间（最直接） #

直接LOAD会把当前时间作为_PARTITIONTIME，不符合需求。正确做法是先把JSONL传到Cloud Storage，再通过INSERT语句映射分区时间：

1. 创建临时外部表：基于Cloud Storage的JSONL文件创建临时表，schema和目标表一致：

CREATE OR REPLACE EXTERNAL TABLE `your-project.your-dataset.temp_old_data`
OPTIONS (
  format = 'NEWLINE_DELIMITED_JSON',
  uris = ['gs://your-bucket/path/to/old-data/*.jsonl']
);

2. 插入数据到目标表：将delivery.timestamp转换为_PARTITIONTIME，同时处理缺失该字段的记录：

INSERT INTO `your-project.your-dataset.target_table`
(mail, delivery, _PARTITIONTIME)
SELECT
  mail,
  delivery,
  TIMESTAMP(delivery.timestamp) AS _PARTITIONTIME
FROM `your-project.your-dataset.temp_old_data`
WHERE delivery.timestamp IS NOT NULL
UNION ALL
SELECT
  mail,
  delivery,
  CURRENT_TIMESTAMP() AS _PARTITIONTIME
FROM `your-project.your-dataset.temp_old_data`
WHERE delivery.timestamp IS NULL;

这个方法不需要修改目标表结构，BigQuery能高效处理200GB级别的数据。

方案2：按批次指定LOAD作业的分区时间 #

如果数据已经按日期分文件存储，可以按批次加载，每个批次指定对应分区时间：

const metadata = {
  sourceFormat: 'NEWLINE_DELIMITED_JSON',
  schema: {
    fields: [
      {name: 'mail', type: 'JSON'},
      {name: 'delivery', type: 'JSON'},
    ],
  },
  timePartitioning: {
    type: 'DAY',
    requirePartitionFilter: false,
  },
  partitionLoadTime: new Date('2016-10-19'), // 该批次数据对应的分区日期
  location: 'US',
  writeDisposition: 'WRITE_APPEND',
};

方案3：切换表的分区类型（长期最优） #

如果可以调整流式管道，建议把目标表改成按delivery.timestamp分区，从根本上解决时间一致性问题：

1. 创建新分区表：

const newTableMetadata = {
  schema: {
    fields: [
      {name: 'mail', type: 'JSON'},
      {name: 'delivery', type: 'JSON'},
    ],
  },
  timePartitioning: {
    type: 'DAY',
    field: 'delivery.timestamp', // 直接用嵌套字段作为分区键
    requirePartitionFilter: true,
  },
  location: 'US',
};

2. 切换流式数据到新表，再导入旧数据
3. 验证无误后，归档或删除旧表

内容的提问来源于stack exchange，提问作者Mr. Demetrius Michael