Escrita em lote e descoberta de registros no Amazon SageMaker Feature Store

O que é o Amazon SageMaker Feature Store

O Amazon SageMaker Feature Store é um repositório gerenciado pela AWS para armazenar, compartilhar e gerenciar features de modelos de Aprendizado de Máquina (ML). Ele oferece serving online de baixa latência para inferência em tempo real, um offline store para retenção histórica e dados de treinamento, além de suporte a padrões de ingestão via streaming e batch.

Os dois problemas que motivaram as novas APIs

À medida que as plataformas de ML amadurecem, dois gaps operacionais aparecem com frequência.

Problema 1: throughput limitado na ingestão

O primeiro desafio envolve pipelines de features de alta vazão. A API existente, o PutRecord, grava um único registro por chamada, em um único feature group. Isso força um padrão de chamadas N×M — N registros multiplicados por M feature groups. O resultado prático: uma pipeline de detecção de fraudes que ingere 10.000 registros por segundo em cinco feature groups precisa sustentar 50.000 chamadas individuais de API por segundo, gerando overhead de conexão e alta latência de cauda.

Problema 2: registros irrecuperáveis no tier In-Memory

O segundo problema afeta quem usa o storage tier In-Memory. Nesse modelo, não há como navegar ou enumerar os registros armazenados no online store. Se os identificadores de registro forem perdidos por um bug ou falha de pipeline, esses registros se tornam permanentemente irrecuperáveis — não existe offline store para consultar, nenhuma query no Amazon Athena para rodar, e nenhuma API para descobrir o que existe.

As duas novas APIs anunciadas

Para endereçar esses dois problemas, a AWS anunciou duas novas APIs para o Amazon SageMaker Feature Store:

• BatchWriteRecord — Grava até 25 registros em múltiplos feature groups em uma única chamada de API, com semântica de sucesso parcial, controle de TTL (Time-to-Live) por registro e as mesmas garantias de ordenação baseadas em EventTime do PutRecord.
• ListRecords — Enumera identificadores de registro dentro de um feature group com paginação. Funciona com os dois tiers de storage: Standard (baseado em Amazon DynamoDB) e In-Memory (baseado em Redis).

Pré-requisitos

Para acompanhar os exemplos apresentados, você precisará de:

• Uma conta AWS com permissões para criar recursos do Amazon SageMaker AI.
• Uma execution role do Amazon SageMaker AI com acesso ao Amazon Simple Storage Service (Amazon S3) e ao AWS Glue, além de permissões para interagir com as APIs do plano de dados do Feature Store. A política mínima de Gerenciamento de Identidade e Acesso (IAM) necessária é:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "sagemaker:BatchWriteRecord",
        "sagemaker:PutRecord",
        "sagemaker:ListRecords"
      ],
      "Resource": "arn:aws:sagemaker:*:*:feature-group/*"
    }
  ]
}

• Boto3 (versão mais recente) ou SageMaker Python SDK v3.8.0 ou superior.
• Um ou mais feature groups existentes com registros ingeridos. Se você é novo no Feature Store, consulte o notebook de workshop end-to-end.

BatchWriteRecord

A API BatchWriteRecord aceita até 25 entradas em uma única requisição, podendo atingir um ou mais feature groups simultaneamente. Cada registro tem sucesso ou falha de forma independente — trata-se de uma API de sucesso parcial, o que significa que falhas individuais não cancelam a requisição inteira.

A API preserva a mesma ordenação baseada em EventTime do PutRecord:

• Se o EventTime do registro recebido for mais recente do que o existente, ele se torna a versão mais atual no online store.
• Caso contrário, o registro é gravado como versão histórica no offline store (para feature groups com storage offline habilitado).
• Registros que falham por outros motivos (erros de autenticação, validação ou throttling) são retornados na resposta com detalhes do erro. As entradas não processadas aparecem em UnprocessedEntries e podem ser retentadas.

Estrutura da requisição

{
  "Entries": [
    {
      "FeatureGroupName": "click-features",
      "Record": [
        {"FeatureName": "user_id", "ValueAsString": "user-123"},
        {"FeatureName": "event_time", "ValueAsString": "2026-06-05T12:00:00Z"},
        {"FeatureName": "click_count", "ValueAsString": "42"}
      ],
      "TargetStores": ["OnlineStore", "OfflineStore"],
      "TtlDuration": {"Unit": "Days", "Value": 7}
    },
    {
      "FeatureGroupName": "login-features",
      "Record": [
        {"FeatureName": "user_id", "ValueAsString": "user-456"},
        {"FeatureName": "event_time", "ValueAsString": "2026-06-05T12:00:01Z"},
        {"FeatureName": "login_count", "ValueAsString": "18"}
      ],
      "TargetStores": ["OnlineStore", "OfflineStore"]
    }
  ]
}

A resposta retorna apenas os registros que falharam. Registros não listados em Errors ou UnprocessedEntries foram gravados com sucesso. A aplicação deve retentar apenas os registros falhos usando exponential backoff para erros retriáveis.

Exemplo de código: ingestão em lote com Boto3

import boto3

featurestore_runtime = boto3.client("sagemaker-featurestore-runtime")

response = featurestore_runtime.batch_write_record(
    Entries=[
        {
            "FeatureGroupName": "click-features",
            "Record": [
                {"FeatureName": "user_id", "ValueAsString": "user-123"},
                {"FeatureName": "event_time", "ValueAsString": "2026-06-05T12:00:00Z"},
                {"FeatureName": "click_count", "ValueAsString": "42"},
            ],
            "TargetStores": ["OnlineStore", "OfflineStore"],
        },
        {
            "FeatureGroupName": "login-features",
            "Record": [
                {"FeatureName": "user_id", "ValueAsString": "user-456"},
                {"FeatureName": "event_time", "ValueAsString": "2026-06-05T12:00:01Z"},
                {"FeatureName": "login_count", "ValueAsString": "18"},
            ],
            "TargetStores": ["OnlineStore", "OfflineStore"],
        },
    ]
)

if response["Errors"]:
    for error in response["Errors"]:
        print(f"Record {error['Entry']}, ErrorCode: {error['ErrorCode']} Failed: {error['ErrorMessage']}")

if response["UnprocessedEntries"]:
    for unprocessed in response["UnprocessedEntries"]:
        print(f"Unprocessed: {unprocessed['FeatureGroupName']}")

if not response["Errors"] and not response["UnprocessedEntries"]:
    print("All records written successfully.")

Exemplo de código: gravando em múltiplos feature groups

É possível atingir múltiplos feature groups em uma única requisição. Os registros são agrupados por feature group e processados de forma independente:

featurestore_runtime = boto3.client("sagemaker-featurestore-runtime")

response = featurestore_runtime.batch_write_record(
    Entries=[
        {
            "FeatureGroupName": "user-profile-features",
            "Record": [
                {"FeatureName": "user_id", "ValueAsString": "user-123"},
                {"FeatureName": "event_time", "ValueAsString": "2026-06-05T12:00:00Z"},
                {"FeatureName": "age", "ValueAsString": "34"},
                {"FeatureName": "region", "ValueAsString": "us-west-2"},
            ],
            "TargetStores": ["OnlineStore"],
        },
        {
            "FeatureGroupName": "click-features",
            "Record": [
                {"FeatureName": "user_id", "ValueAsString": "user-123"},
                {"FeatureName": "event_time", "ValueAsString": "2026-06-05T12:00:00Z"},
                {"FeatureName": "click_count", "ValueAsString": "42"},
            ],
            "TargetStores": ["OnlineStore", "OfflineStore"],
        },
    ]
)

Uma falha em um feature group não afeta os registros destinados a outros feature groups.

Suporte a TTL (Time-to-Live)

A API BatchWriteRecord suporta TTL em três níveis de precedência:

• TTL por registro — Definido via TtlDuration em entradas individuais. Tem a maior prioridade.
• TTL por requisição — Um TtlDuration padrão no nível superior da requisição, aplicado às entradas sem TTL de registro.
• TTL por feature group — O TTL configurado no próprio feature group, aplicado quando nem o TTL de registro nem o de requisição estão definidos.

Considerações importantes

• Máximo de 25 entradas por requisição, contando o total de entradas em todos os feature groups.
• Semântica de sucesso parcial: ao contrário de APIs transacionais, o BatchWriteRecord não desfaz gravações bem-sucedidas se alguns registros falharem. Projete a lógica de retry para reenviar apenas os registros retornados em Errors.
• Modelo de IAM similar ao PutRecord: o chamador precisa ter permissões sagemaker:BatchWriteRecord e sagemaker:PutRecord no ARN (Nome de Recurso da Amazon) de cada feature group alvo.
• Ordenação por EventTime preservada: um registro desatualizado não pode sobrescrever um mais recente no online store.
• Flexibilidade de TargetStores: cada entrada pode independentemente ter como alvo o OnlineStore, o OfflineStore ou ambos.

ListRecords

A API ListRecords fecha a lacuna na descoberta de registros para ambos os tiers de storage. Ela enumera identificadores de registro dentro de um feature group usando paginação, retornando apenas registros ativos, não deletados e não expirados — prontos para uso com GetRecord ou DeleteRecord.

O problema com a descoberta de registros

O Feature Store suporta PutRecord, GetRecord e DeleteRecord, mas todas exigem que o chamador conheça o identificador exato do registro. Não havia API para navegar ou enumerar registros dentro de um feature group.

Para o tier Standard, o workaround era consultar o offline store via Amazon Athena — o que exige configuração de offline store, adiciona custo e não é em tempo real. Para o tier In-Memory, a situação é crítica: não há offline store correspondente por padrão. Se os identificadores de registro forem perdidos, esses registros se tornam completamente irrecuperáveis, gerando dados fantasmas, custos de storage desperdiçados e riscos de conformidade quando usuários solicitam exclusão de dados.

Como o ListRecords funciona

A API funciona com os dois tiers de storage:

• Tier Standard (Amazon DynamoDB): Faz scan do online store, retornando o identificador da versão mais recente de cada registro. Registros soft-deleted e expirados são automaticamente excluídos.
• Tier In-Memory (Redis): Faz scan das keys e filtra registros soft-deleted e keys internas do sistema. Retorna identificadores de registro extraídos dos nomes das keys.

Exemplo de código: enumerar todos os registros de um feature group

import boto3

featurestore_runtime = boto3.client("sagemaker-featurestore-runtime")

all_identifiers = []
next_token = None

while True:
    params = {
        "FeatureGroupName": "user-profile-features",
        "MaxResults": 100,
    }
    if next_token:
        params["NextToken"] = next_token

    response = featurestore_runtime.list_records(**params)
    all_identifiers.extend(response["RecordIdentifiers"])
    next_token = response.get("NextToken")

    if not next_token:
        break

print(f"Found {len(all_identifiers)} active records.")

Exemplo de código: limpeza de registros órfãos

Um caso de uso comum é identificar e deletar registros que não são mais necessários. Isso é crítico para feature groups do tier In-Memory, onde registros órfãos persistem indefinidamente:

import boto3

featurestore_runtime = boto3.client("sagemaker-featurestore-runtime")

# Step 1: Enumerate all record identifiers
all_ids = []
next_token = None

while True:
    params = {"FeatureGroupName": "session-features", "MaxResults": 100}
    if next_token:
        params["NextToken"] = next_token

    response = featurestore_runtime.list_records(**params)
    all_ids.extend(response["RecordIdentifiers"])
    next_token = response.get("NextToken")

    if not next_token:
        break

# Step 2: Compare against your application's active session list
active_sessions = get_active_sessions()  # Your application logic
orphaned = [rid for rid in all_ids if rid not in active_sessions]

# Step 3: Delete orphaned records
for record_id in orphaned:
    featurestore_runtime.delete_record(
        FeatureGroupName="session-features",
        RecordIdentifierValueAsString=record_id,
        EventTime="2026-06-05T12:00:00Z",
    )

print(f"Deleted {len(orphaned)} orphaned records.")

Comportamento da paginação

• Tamanho da página: configurável via MaxResults (padrão 10, máximo 100).
• Formato do token: string opaca e criptografada. Não tente parsear ou construir tokens — passe-os sem alteração.
• Ordenação: os resultados não têm garantia de ordem específica.
• Escritas concorrentes: se registros forem gravados ou deletados durante a paginação, podem ocorrer duplicatas ou lacunas. Esse é um comportamento documentado.
• Escopo do token: tokens estão vinculados a um feature group e conta específicos e não podem ser reutilizados entre eles.

Considerações importantes

• Apenas identificadores de registro: a versão atual retorna identificadores de registro sem valores de features. Use GetRecord ou BatchGetRecord para recuperar os registros completos.
• Filtragem automática: a API exclui registros soft-deleted, expirados e keys internas do sistema. Você vê apenas registros ativos e recuperáveis.
• Permissão IAM: o chamador precisa ter a permissão sagemaker:ListRecords no ARN do feature group.
• Ambos os tiers suportados: o ListRecords funciona de forma idêntica do ponto de vista do chamador, independentemente do tier de storage.

Combinando as duas APIs: um caso de uso completo

As duas APIs se complementam naturalmente. Veja um exemplo de workflow de conformidade que verifica a exclusão completa de dados de um usuário em múltiplos feature groups:

import boto3

featurestore_runtime = boto3.client("sagemaker-featurestore-runtime")

feature_groups = ["user-profiles", "click-history", "purchase-signals"]
user_to_delete = "user-789"

# Step 1: Find and delete the user across all feature groups
for fg_name in feature_groups:
    all_ids = []
    next_token = None

    while True:
        params = {"FeatureGroupName": fg_name, "MaxResults": 100}
        if next_token:
            params["NextToken"] = next_token

        response = featurestore_runtime.list_records(**params)
        all_ids.extend(response["RecordIdentifiers"])
        next_token = response.get("NextToken")

        if not next_token:
            break

    if user_to_delete in all_ids:
        featurestore_runtime.delete_record(
            FeatureGroupName=fg_name,
            RecordIdentifierValueAsString=user_to_delete,
            EventTime="2026-06-05T23:59:59Z",
        )
        print(f"Deleted '{user_to_delete}' from {fg_name}")

# Step 2: Log the deletion event using BatchWriteRecord
featurestore_runtime.batch_write_record(
    Entries=[
        {
            "FeatureGroupName": "deletion-audit-log",
            "Record": [
                {"FeatureName": "request_id", "ValueAsString": "del-001"},
                {"FeatureName": "event_time", "ValueAsString": "2026-06-05T23:59:59Z"},
                {"FeatureName": "user_id", "ValueAsString": user_to_delete},
                {"FeatureName": "status", "ValueAsString": "completed"},
                {"FeatureName": "feature_groups_cleaned", "ValueAsString": "3"},
            ],
            "TargetStores": ["OnlineStore", "OfflineStore"],
        }
    ]
)

Limpeza de recursos

Para evitar cobranças contínuas, exclua os feature groups criados durante os testes. Para feature groups do tier In-Memory, use o ListRecords para enumerar os registros e o DeleteRecord para removê-los antes de excluir o feature group.

Conclusão

As APIs BatchWriteRecord e ListRecords representam melhorias significativas no plano de dados do Amazon SageMaker Feature Store. O BatchWriteRecord reduz o volume de chamadas de API para ingestão de alta vazão em até 25x, preservando as garantias de ordenação baseadas em EventTime. O ListRecords habilita a descoberta e o gerenciamento do ciclo de vida de registros — algo crítico para clientes do tier In-Memory, que anteriormente não tinham como enumerar ou limpar seus dados.

Juntas, essas APIs viabilizam padrões que antes eram difíceis ou impossíveis: pipelines de ingestão em bulk com menos conexões e menor latência, workflows de conformidade que verificam exclusão completa de dados, e ferramentas operacionais que navegam pelo conteúdo de feature groups em tempo real.

Para mais informações, consulte a documentação do Feature Store, a referência de API do Feature Store, a documentação de configuração do offline store e o anúncio de novidades. Para aprofundar o conhecimento sobre as capacidades do Feature Store, confira também os posts relacionados: Understanding the Key Capabilities of Amazon SageMaker Feature Store, Accelerate ML Feature Pipelines with new capabilities in Amazon SageMaker Feature Store e Using Streaming Ingestion with Amazon SageMaker Feature Store.

Fonte

Batch write and discover records in Amazon SageMaker Feature Store (https://aws.amazon.com/blogs/machine-learning/batch-write-and-discover-records-in-amazon-sagemaker-feature-store/)