Em 22 de novembro de 2024, ingestão do Amazon OpenSearch lançado suporte para processadores AWS Lambda. Com este lançamento, você agora tem mais flexibilidade para enriquecer e transformar seus logs, métricas e dados de rastreamento em um pipeline de ingestão do OpenSearch. Alguns exemplos incluem o uso de modelos básicos (FMs) para gerar incorporações vetoriais para seus dados e procurar fontes de dados externas, como Amazon DynamoDB para enriquecer seus dados.
Ingestão do Amazon OpenSearch é um pipeline de dados totalmente gerenciado e sem servidor que fornece dados de log, métricas e rastreamento em tempo actual para Serviço Amazon OpenSearch domínios e Amazon OpenSearch sem servidor coleções.
Processadores são componentes em um pipeline do OpenSearch Ingestion que permitem filtrar, transformar e enriquecer eventos usando o formato desejado antes de publicar registros em um destino de sua escolha. Se nenhum processador for definido na configuração do pipeline, os eventos serão publicados no formato especificado pelo componente de origem. Você pode incorporar vários processadores em um único pipeline e eles serão executados sequencialmente conforme definido na configuração do pipeline.
O OpenSearch Ingestion oferece a opção de usar funções Lambda como processadores junto com processadores nativos integrados ao transformar dados. Você pode agrupar eventos em um único payload com base na contagem ou tamanho do evento antes de invocar o Lambda para otimizar o pipeline em termos de desempenho e custo. O Lambda permite executar código sem provisionar ou gerenciar servidores, eliminando a necessidade de criar lógica de escalabilidade de cluster com reconhecimento de carga de trabalho, manter integrações de eventos ou gerenciar tempos de execução.
Nesta postagem, demonstramos como usar o processador Lambda do OpenSearch Ingestion para gerar embeddings para seus dados de origem e ingeri-los em um Coleção de vetores OpenSearch Serverless. Esta solução usa a flexibilidade dos pipelines de ingestão do OpenSearch com um processador Lambda para gerar embeddings dinamicamente. A função Lambda invocará o Modelo de incorporação de texto Amazon Titan hospedado em Base Amazônicapermitindo a criação de incorporação eficiente e escalonável. Essa arquitetura simplifica vários casos de uso, incluindo mecanismos de recomendação, chatbots personalizados e sistemas de detecção de fraudes.
A integração do OpenSearch Ingestion, Lambda e OpenSearch Serverless cria um pipeline totalmente sem servidor para incorporação de geração e pesquisa. Essa combinação oferece escalonamento automático para atender às demandas da carga de trabalho e um modelo orientado ao uso. As operações são simplificadas porque a AWS gerencia infraestrutura, atualizações e manutenção. Essa abordagem sem servidor permite que você se concentre no desenvolvimento de soluções de pesquisa e análise, em vez de no gerenciamento da infraestrutura.
Observe que o Amazon OpenSearch Service também fornece Pesquisa neural que transforma texto em vetores e facilita a busca de vetores tanto no momento da ingestão quanto no momento da busca. Durante a ingestão, a pesquisa neural transforma o texto do documento em incorporações vetoriais e indexa o texto e suas incorporações vetoriais em um índice vetorial. A pesquisa neural é disponível para clusters gerenciados que executam a versão 2.9 e superior.
Visão geral da solução
Esta solução cria embeddings em um conjunto de dados armazenado em Serviço de armazenamento simples da Amazon (Amazon S3). Usamos a função Lambda para invocar o modelo Amazon Titan na carga entregue pelo OpenSearch Ingestion.
Pré-requisitos
Você deve ter uma função apropriada com permissões para invocar sua função Lambda e modelo Amazon Bedrock e também gravar na coleção OpenSearch Serverless.
Para fornecer acesso à coleção, você deve configurar um Gerenciamento de identidade e acesso da AWS (IAM) com uma política de permissões que concede acesso à coleção. Para mais detalhes, consulte Conceder acesso aos pipelines de ingestão do Amazon OpenSearch às coleções. A seguir está o código de exemplo:
{
"Model": "2012-10-17",
"Assertion": (
{
"Sid": "allowinvokeFunction",
"Impact": "Enable",
"Motion": (
"lambda:InvokeFunction"
),
"Useful resource": "arn:aws:lambda:{{area}}:{{account-id}}:perform:{{function-name}}"
}
)
}
A função deve ter a seguinte relação de confiança, o que permite que o OpenSearch Ingestion assuma-a:
{
"Model": "2012-10-17",
"Assertion": (
{
"Impact": "Enable",
"Principal": {
"Service": "osis-pipelines.amazonaws.com"
},
"Motion": "sts:AssumeRole"
}
)
}
Criar um pipeline de ingestão
Você pode criar um pipeline usando um blueprint. Para esta postagem, selecionamos o Enriquecimento personalizado do AWS Lambda projeto.
Nós usamos o Conjunto de dados básicos de títulos IMDBque contém informações do filme, incluindo originalTitle, runtimeMinutese gêneros.
O pipeline do OpenSearch Ingestion usa um processador Lambda para criar embeddings para o campo original_title e armazene os embeddings como original_title_embeddings juntamente com outros dados.
Veja o seguinte código do pipeline:
Vamos dar uma olhada mais de perto no processador Lambda no pipeline de ingestão. Preste atenção ao key_nameparâmetro. Você pode escolher qualquer valor para key_name e sua função Lambda precisará fazer referência a essa chave em sua função Lambda ao processar a carga útil do OpenSearch Ingestion. O tamanho da carga útil é determinado pela configuração do lote. Quando o lote é habilitado no processador Lambda, o OpenSearch Ingestion agrupa vários eventos em uma única carga antes de invocar a função Lambda. Um lote é enviado ao Lambda quando qualquer um dos seguintes limites é atingido:
- contagem_de_eventos – O número de eventos atinge o limite especificado
- tamanho_máximo – O tamanho whole do lote atinge o tamanho especificado (por exemplo, 5 MB) e é configurável até 6 MB (limite de carga útil de invocação para AWS Lambda)
Função lambda
A função Lambda recebe os dados do OpenSearch Ingestion, invoca o Amazon Bedrock para gerar a incorporação e os adiciona ao registro de origem. “paperwork”é usado para fazer referência aos eventos provenientes do OpenSearch Ingestion e corresponde ao key_name declarado no pipeline. Adicionamos a incorporação do Amazon Bedrock de volta ao registro unique. Esse novo registro com o valor de incorporação anexado é então enviado ao coletor OpenSearch Serverless pelo OpenSearch Ingestion. Veja o seguinte código:
import json
import boto3
import os
# Initialize Bedrock consumer
bedrock = boto3.consumer('bedrock-runtime')
def generate_embedding(textual content):
"""Generate embedding for the given textual content utilizing Bedrock."""
response = bedrock.invoke_model(
modelId="amazon.titan-embed-text-v1",
contentType="software/json",
settle for="software/json",
physique=json.dumps({"inputText": textual content})
)
embedding = json.hundreds(response('physique').learn())('embedding')
return embedding
def lambda_handler(occasion, context):
# Assuming the enter is an inventory of JSON paperwork
paperwork = occasion('paperwork')
processed_documents = ()
for doc in paperwork:
if originalTitle' in doc:
# Generate embedding for the 'originalTitle' area
embedding = generate_embedding(doc(originalTitle'))
# Add the embedding to the doc
doc('originalTitle_embeddings') = embedding
processed_documents.append(doc)
# Return the processed paperwork
return processed_documents
No caso de qualquer exceção ao usar o processador lambda, todos os documentos do lote são considerados eventos com falha e são encaminhados para a próxima cadeia de processadores, se houver, ou para o coletor com uma tag de falha. A tag pode ser configurada para o pipeline com o tags_on_failure parâmetro e os erros também são enviados aos logs do CloudWatch para ações adicionais.
Após a execução do pipeline, você poderá ver que os embeddings foram criados e armazenados como originalTitle_embeddings dentro do documento em um índice k-NN, imdb-data-embeddings. A captura de tela a seguir mostra um exemplo.
Resumo
Nesta postagem, mostramos como você pode usar o Lambda como parte do pipeline do OpenSearch Ingestion para permitir transformação complexa e enriquecimento de seus dados. Para obter mais detalhes sobre o recurso, consulte Usar um pipeline de ingestão do OpenSearch com AWS Lambda.
Sobre os Autores
Jagadish Kumar (Jag) é arquiteto de soluções especialista sênior na AWS com foco no Amazon OpenSearch Service. Ele é profundamente apaixonado por arquitetura de dados e ajuda os clientes a criar soluções analíticas em escala na AWS.
Sam Selvan é arquiteto de soluções especialista principal do Amazon OpenSearch Service.
Srikanth Govindarajan é engenheiro de desenvolvimento de software program no Amazon Opensearch Service. Srikanth é apaixonado por arquitetar infraestrutura e construir soluções escaláveis para casos de uso baseados em pesquisa, análise, segurança, IA e aprendizado de máquina.