用无服务器架构查询基因组变异：拆解 CSIRO 的 AWS sBeacon 方案

2026-09-18 31 预计阅读时间: 1 分钟
来源: aws.amazon.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间：11 分钟

基因组变异查询看起来像一次普通的数据检索，真正进入临床和科研环境后，却要同时面对海量数据、突发流量、访问控制和成本约束。澳大利亚国家科学机构 CSIRO 构建的 Serverless Beacon（sBeacon），通过 Amazon S3、AWS Lambda、Amazon DynamoDB 和 Amazon Athena 实现 GA4GH Beacon 标准，为生产规模的基因组变异查询提供了一种无服务器路径。

这套设计值得关注的地方，不只是“没有服务器要维护”，而是把在线查询、对象存储和分析任务分给不同服务，避免让一个数据库承担所有工作。

把不同查询交给不同的数据层

GA4GH Beacon 的核心场景，是让调用方按照染色体位置、参考碱基和替代碱基等条件，询问某个数据集中是否存在对应变异。在真实系统中，查询通常可以分成两类：

  • 低延迟、结构明确的在线查询：例如根据标准化后的变异键判断记录是否存在。
  • 范围更大、条件更复杂的分析查询：例如研究人员对存放在对象存储中的数据执行聚合、过滤或队列分析。

结合 sBeacon 使用的 AWS 服务，可以把职责理解为：

  • Amazon S3 保存大规模基因组变异数据和分析文件，适合承载 Parquet 等列式格式。
  • AWS Lambda 负责校验 Beacon 请求、规范化变异参数、执行授权并组织响应。
  • Amazon DynamoDB 支撑可预测键值上的低延迟读取，例如精确变异查询或元数据索引。
  • Amazon Athena 直接查询 S3 数据，用于不适合建成在线键值索引的复杂分析。

这里的关键不是把每一份数据同时复制到所有服务，而是先识别访问模式：高频精确查询进入 DynamoDB，体量大但频率较低的分析留在 S3，并由 Athena 按需扫描。

这种拆分也直接影响成本。Lambda 按调用和执行资源计费，DynamoDB 可以使用按需容量，Athena 按扫描数据量计费，S3 则承担相对经济的持久化存储。系统不需要为了偶发峰值长期运行固定规模的计算集群。

一个可改造的最小查询服务

下面是一个简化的实践样例，用于展示如何把 Beacon 风格的精确变异查询映射到 Lambda 和 DynamoDB。它不是 sBeacon 的原始代码，也没有覆盖完整 GA4GH Beacon 规范；假设已经将变异规范化为以下键：

{referenceName}:{start}:{referenceBases}:{alternateBases}

运行前需要安装并配置 AWS CLI 与 AWS SAM CLI。创建如下目录：

beacon-demo/
├── template.yaml
└── src/
    └── app.py

template.yaml：

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Minimal serverless genomic variant lookup API

Globals:
  Function:
    Runtime: python3.12
    Timeout: 10
    MemorySize: 256

Resources:
  BeaconApi:
    Type: AWS::Serverless::HttpApi

  VariantTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      SSESpecification:
        SSEEnabled: true
      AttributeDefinitions:
        - AttributeName: variant_key
          AttributeType: S
      KeySchema:
        - AttributeName: variant_key
          KeyType: HASH

  VariantDataBucket:
    Type: AWS::S3::Bucket
    Properties:
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true
      BucketEncryption:
        ServerSideEncryptionConfiguration:
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: AES256

  QueryFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: src/
      Handler: app.handler
      Environment:
        Variables:
          VARIANT_TABLE: !Ref VariantTable
      Policies:
        - DynamoDBReadPolicy:
            TableName: !Ref VariantTable
      Events:
        QueryVariant:
          Type: HttpApi
          Properties:
            ApiId: !Ref BeaconApi
            Path: /variants
            Method: GET

Outputs:
  ApiUrl:
    Value: !Sub 'https://${BeaconApi}.execute-api.${AWS::Region}.${AWS::URLSuffix}/variants'
  VariantTableName:
    Value: !Ref VariantTable
  VariantBucketName:
    Value: !Ref VariantDataBucket

src/app.py：

import json
import os

import boto3

TABLE = boto3.resource("dynamodb").Table(os.environ["VARIANT_TABLE"])
REQUIRED = ("referenceName", "start", "referenceBases", "alternateBases")


def response(status_code, body):
    return {
        "statusCode": status_code,
        "headers": {"content-type": "application/json"},
        "body": json.dumps(body),
    }


def handler(event, context):
    params = event.get("queryStringParameters") or {}
    missing = [name for name in REQUIRED if params.get(name) in (None, "")]
    if missing:
        return response(400, {"error": "missing query parameters", "fields": missing})

    try:
        start = int(params["start"])
        if start < 0:
            raise ValueError
    except ValueError:
        return response(400, {"error": "start must be a non-negative integer"})

    variant_key = ":".join(
        [
            params["referenceName"],
            str(start),
            params["referenceBases"].upper(),
            params["alternateBases"].upper(),
        ]
    )

    item = TABLE.get_item(
        Key={"variant_key": variant_key},
        ProjectionExpression="variant_key",
        ConsistentRead=False,
    ).get("Item")

    return response(
        200,
        {
            "exists": item is not None,
            "query": {
                "referenceName": params["referenceName"],
                "start": start,
                "referenceBases": params["referenceBases"].upper(),
                "alternateBases": params["alternateBases"].upper(),
            },
        },
    )

部署并取得输出：

cd beacon-demo
sam build
sam deploy --guided
aws cloudformation describe-stacks \
  --stack-name beacon-demo \
  --query 'Stacks[0].Outputs' \
  --output table

部署完成后，可以先向 DynamoDB 写入测试变异。将表名替换为 CloudFormation 输出值：

TABLE_NAME='替换为表名'

aws dynamodb put-item \
  --table-name "$TABLE_NAME" \
  --item '{"variant_key":{"S":"chr1:100:A:G"}}'

然后将 URL 替换为 ApiUrl 输出值：

API_URL='替换为ApiUrl'

curl -G "$API_URL" \
  --data-urlencode 'referenceName=chr1' \
  --data-urlencode 'start=100' \
  --data-urlencode 'referenceBases=A' \
  --data-urlencode 'alternateBases=G'

预期会得到类似响应：

{"exists": true, "query": {"referenceName": "chr1", "start": 100, "referenceBases": "A", "alternateBases": "G"}}

生产系统还需要处理参考基因组版本、染色体命名差异、变异左对齐与规范化、多等位基因拆分等问题。否则，同一个生物学变异可能生成不同的键，导致错误的阴性结果。

Athena 适合承接哪些任务

DynamoDB 适合已知键的快速读取，但不适合随意增加过滤条件。对于队列统计、数据质量检查或研究型探索，可以将规范化结果以 Parquet 格式写入 S3，再通过 Athena 查询。

下面的 SQL 只是一个可改造的示例，字段和 S3 路径需要按照实际数据布局调整：

CREATE EXTERNAL TABLE IF NOT EXISTS genomics.variants (
  position BIGINT,
  reference_bases STRING,
  alternate_bases STRING,
  sample_count BIGINT
)
PARTITIONED BY (
  assembly STRING,
  reference_name STRING
)
STORED AS PARQUET
LOCATION 's3://YOUR-BUCKET/variants/';

SELECT reference_name, COUNT(*) AS variant_count
FROM genomics.variants
WHERE assembly = 'GRCh38'
  AND reference_name = 'chr1'
  AND position BETWEEN 1000000 AND 2000000
GROUP BY reference_name;

为了控制 Athena 成本，应优先采用列式格式、压缩数据，并按照常用过滤字段设计分区。还可以配置 Athena workgroup 的扫描量限制，防止错误查询扫描整个数据湖。

在线 API 不应在每次精确查询时无条件启动 Athena。Athena 的优势是灵活扫描和分析，而不是替代低延迟索引。更稳妥的做法是让批处理流程生成 DynamoDB 查询索引，同时保留 S3 中的完整分析数据。

基因组查询的安全边界

“是否存在某个变异”看似只返回布尔值，但连续、有针对性的查询仍可能泄露敏感信息。无服务器架构会减少基础设施运维工作，却不会自动解决数据治理问题。

进入临床或受监管环境前，至少要检查：

  • API 是否要求身份认证，并按数据集、项目或机构执行授权。
  • 是否设置查询速率限制、异常模式检测和审计日志。
  • Lambda 角色是否只拥有必要的 DynamoDB、S3 和 Athena 权限。
  • S3、DynamoDB、日志以及查询结果是否使用符合要求的加密密钥。
  • Beacon 响应是否需要阈值抑制、结果模糊化或审批机制。
  • Athena 查询结果桶是否阻止公共访问，并设置生命周期策略。
  • 日志中是否意外记录样本标识、完整查询参数或其他敏感数据。

对于高敏感度数据，还应考虑私有网络连接、组织级服务控制策略、跨账户隔离以及针对数据驻留要求的区域选择。

采用这类架构时如何取舍

sBeacon 展示了一条清晰路线：用 GA4GH Beacon 提供互操作接口，用 Lambda 承载无状态计算，用 DynamoDB 响应可预测的在线访问，用 S3 和 Athena保留大规模分析能力。

落地时可以按以下顺序推进：

  1. 先定义变异规范化规则和实际查询模式，再设计键和分区。
  2. 将最常见的精确查询放入 DynamoDB，不要把所有分析维度都做成在线索引。
  3. 使用 Parquet、压缩和合理分区降低 Athena 扫描成本。
  4. 在性能测试之外增加授权绕过、查询枚举和日志泄露测试。
  5. 通过调用次数、Lambda 延迟、DynamoDB 消耗和 Athena 扫描字节建立成本看板。

无服务器并不意味着容量设计消失，而是容量问题转化为索引设计、并发限制、数据布局和单次查询成本。对于基因组数据，这种拆分尤其重要：把快速回答与深度分析分开，才能同时守住延迟、费用和安全边界。

# AWS # 云平台 # 后端 # 架构设计

相关推荐