如何通过CloudFormation在空Timestream表上部署定时查询？

问题背景 #

使用AWS SAM部署CloudFormation栈时，会创建Timestream数据库、表以及定时查询。但Timestream采用动态Schema机制，只有表中写入数据后，对应列才会被系统识别。初始部署阶段表为空，定时查询SQL中引用的dev_eui、state等列不存在，导致CloudFormation验证查询时报错：

Resource handler returned message: "line 3:8: Column 'dev_eui' does not exist (Service: AmazonTimestreamQuery; Status Code: 400; Error Code: ValidationException;

当前采用两个独立SAM模板分步部署的方式，以下是更简洁的替代方案：

方案1：修改SQL语句，兼容空表Schema验证 #

Timestream Query支持TRY()函数，当列不存在时会返回NULL而非抛出错误，修改查询语句即可让CloudFormation验证通过：

修改后的SQL示例：

WITH raw_data as (
    SELECT time
        ,TRY(dev_eui) as dev_eui
        ,TRY(channel) as channel
        ,min(TRY(state)) as state
    FROM MachineMonitoring.Readings
    WHERE time BETWEEN @scheduled_runtime - 1h AND @scheduled_runtime
    GROUP BY time, TRY(dev_eui), TRY(channel)
), ...

• 用TRY()包裹所有引用的动态列，部署时即使列不存在，SQL语法验证也能顺利通过
• 当表中有实际数据写入后，TRY()会正常返回列值，完全不影响查询逻辑

方案2：用自定义资源Lambda预写入测试数据 #

在CloudFormation栈中添加Lambda自定义资源，在表创建完成后自动写入一条测试数据，让Timestream生成所需Schema，再部署定时查询：

1. 在SAM模板中定义相关资源：

Resources:
  # 定义Timestream表
  ReadingsTable:
    Type: AWS::Timestream::Table
    Properties:
      DatabaseName: MachineMonitoring
      TableName: Readings

  # 自定义资源Lambda，负责写入测试数据
  SeedTimestreamDataFunction:
    Type: AWS::Serverless::Function
    Properties:
      Runtime: python3.11
      Handler: index.lambda_handler
      CodeUri: ./seed-data/
      Policies:
        - AmazonTimestreamFullAccess
      DependsOn: ReadingsTable

  # 触发自定义资源执行
  SeedDataResource:
    Type: Custom::SeedTimestreamData
    Properties:
      ServiceToken: !GetAtt SeedTimestreamDataFunction.Arn
      DatabaseName: MachineMonitoring
      TableName: Readings

  # 定时查询依赖自定义资源，确保数据已写入、Schema已生成
  HourlyAggregationQuery:
    Type: AWS::Timestream::ScheduledQuery
    Properties:
      QueryString: !Ref AggregationQuerySQL
      ScheduleConfiguration:
        ScheduleExpression: "cron(0 * * * ? *)"
      # 其他配置（如目标存储、IAM角色等）...
    DependsOn: SeedDataResource

2. Lambda函数代码（seed-data/index.py）：

import boto3
import json

timestream_write = boto3.client('timestream-write')

def lambda_handler(event, context):
    db_name = event['ResourceProperties']['DatabaseName']
    table_name = event['ResourceProperties']['TableName']
    
    try:
        # 写入测试数据，覆盖所需列
        timestream_write.write_records(
            DatabaseName=db_name,
            TableName=table_name,
            Records=[
                {
                    'Time': '2024-01-01T00:00:00.000000Z',
                    'Dimensions': [
                        {'Name': 'dev_eui', 'Value': '6081F9xxxxxxxxxx'},
                        {'Name': 'channel', 'Value': '1'}
                    ],
                    'MeasureName': 'reading',
                    'MeasureValue': '0.0',
                    'MeasureValueType': 'DOUBLE',
                    'TimeUnit': 'MILLISECONDS'
                },
                {
                    'Time': '2024-01-01T00:00:00.000000Z',
                    'Dimensions': [
                        {'Name': 'dev_eui', 'Value': '6081F9xxxxxxxxxx'},
                        {'Name': 'channel', 'Value': '1'}
                    ],
                    'MeasureName': 'state',
                    'MeasureValue': '0',
                    'MeasureValueType': 'BIGINT',
                    'TimeUnit': 'MILLISECONDS'
                }
            ]
        )
        return {'Status': 'SUCCESS', 'PhysicalResourceId': f"{db_name}-{table_name}-seeded"}
    except Exception as e:
        return {'Status': 'FAILED', 'Reason': str(e)}

• 自定义资源会在表创建后立即写入测试数据，生成所需的列Schema
• 定时查询依赖该自定义资源，确保验证时列已存在

方案3：用SAM参数+Condition控制定时查询部署 #

通过参数控制是否部署定时查询，第一次部署时跳过，待数据写入后再更新部署开启：

1. 在SAM模板中添加参数和Condition：

Parameters:
  DeployScheduledQuery:
    Type: String
    Default: "false"
    AllowedValues: ["true", "false"]

Conditions:
  ShouldDeployScheduledQuery: !Equals [!Ref DeployScheduledQuery, "true"]

Resources:
  # Timestream数据库和表定义...

  # 定时查询仅在条件满足时部署
  HourlyAggregationQuery:
    Type: AWS::Timestream::ScheduledQuery
    Condition: ShouldDeployScheduledQuery
    Properties:
      QueryString: !Ref AggregationQuerySQL
      ScheduleConfiguration:
        ScheduleExpression: "cron(0 * * * ? *)"
      # 其他配置...

2. 部署步骤：

• 第一次部署：执行sam deploy --parameter-overrides DeployScheduledQuery=false，仅创建数据库和表
• 手动或通过脚本写入测试数据到Readings表
• 更新部署：执行sam deploy --parameter-overrides DeployScheduledQuery=true，创建定时查询

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