Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.
# Carga de datos de streaming en Amazon OpenSearch Service
Puedes usar OpenSearch Ingestion para cargar directamente [los datos de streaming](https://aws.amazon.com/streaming-data/) en tu dominio de Amazon OpenSearch Service, sin necesidad de utilizar soluciones de terceros. Para enviar datos a OpenSearch Ingestion, debes configurar tus generadores de datos y el servicio entrega automáticamente los datos al dominio o la colección que especifiques. Para empezar a usar OpenSearch Ingestion, consulte. [Tutorial: Ingerir datos en una colección mediante Amazon OpenSearch Ingestion](osis-serverless-get-started.md)
Puedes seguir utilizando otras fuentes para cargar datos de streaming, como Amazon Data Firehose y Amazon CloudWatch Logs, que tienen soporte integrado para OpenSearch Service. Otros, como Amazon S3, Amazon Kinesis Data Streams y Amazon DynamoDB, utilizan funciones AWS Lambda como controladores de eventos. Las funciones de Lambda responden a los nuevos datos procesándolos y transmitiéndolos al dominio.
**nota**
Lambda es compatible con varios lenguajes de programación populares y está disponible en la mayoría de las Regiones de AWS. Para más información, consulte [Introducción a Lambda](https://docs.aws.amazon.com/lambda/latest/dg/lambda-app.html) en la *Guía para desarrolladores de AWS Lambda * y [Puntos de conexión de servicio de AWS](https://docs.aws.amazon.com/general/latest/gr/rande.html#lambda_region) en *Referencia general de AWS*.
# Cargando datos de streaming desde Ingestion OpenSearch
Puede usar Amazon OpenSearch Ingestion para cargar datos en un dominio OpenSearch de servicio. Usted configura sus generadores de datos para que envíen datos a OpenSearch Ingestion y esta entrega automáticamente los datos a la colección que usted especifique. También puede configurar OpenSearch Ingestion para transformar los datos antes de entregarlos. Para obtener más información, consulte [Información general sobre Amazon OpenSearch Ingestion](ingestion.md).
# Carga de datos de streaming desde Amazon S3
Puede usar Lambda para enviar datos a su dominio de OpenSearch servicio desde Amazon S3. Cuando llegan datos nuevos a un bucket de S3, activan una notificación de eventos en Lambda que, a su vez, ejecuta el código personalizado para realizar la indexación.
Este método de streaming de datos es extremadamente flexible. Es posible [indexar los metadatos del objeto](https://aws.amazon.com/blogs/database/indexing-metadata-in-amazon-elasticsearch-service-using-aws-lambda-and-python/) o, si el objeto contiene texto sin formato, analizar e indexar algunos elementos del cuerpo del objeto. En esta sección, se incluye código de muestra sencillo de Python que utiliza expresiones regulares para analizar un archivo de registros e indexar los resultados obtenidos.
## Requisitos previos
Antes de continuar, debe contar con los siguientes recursos.
****
| Requisito previo | Description (Descripción) |
| --- | --- |
| Bucket de Amazon S3 | Para más información, consulte [Crear su primer bucket de S3](https://docs.aws.amazon.com/AmazonS3/latest/userguide/CreatingABucket.html) en la Guía del usuario de Amazon Simple Storage Service. El bucket debe residir en la misma región que su dominio OpenSearch de servicio. |
| OpenSearch Dominio de servicio | Es el destino de los datos después de que la función de Lambda los procesa. Para más información, consulte [Creación de dominios OpenSearch de servicio](createupdatedomains.md#createdomains). |
## Crear el paquete de implementación de Lambda
Los paquetes de implementación son archivos ZIP o JAR que contienen el código y sus dependencias. En esta sección, se incluye un código de muestra de Python. Para otros lenguajes de programación, consulte [Paquetes de implementación de Lambda](https://docs.aws.amazon.com/lambda/latest/dg/gettingstarted-package.html) en la *Guía para desarrolladores de AWS Lambda *.
1. Cree un directorio. En este ejemplo, utilizamos el nombre `s3-to-opensearch`.
1. Cree un archivo en el directorio denominado `sample.py`:
```
import boto3
import re
import requests
from requests_aws4auth import AWS4Auth
region = '' # e.g. us-west-1
service = 'es'
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)
host = '' # the OpenSearch Service domain, e.g. https://search-mydomain.us-west-1.es.amazonaws.com
index = 'lambda-s3-index'
datatype = '_doc'
url = host + '/' + index + '/' + datatype
headers = { "Content-Type": "application/json" }
s3 = boto3.client('s3')
# Regular expressions used to parse some simple log lines
ip_pattern = re.compile('(\d+\.\d+\.\d+\.\d+)')
time_pattern = re.compile('\[(\d+\/\w\w\w\/\d\d\d\d:\d\d:\d\d:\d\d\s-\d\d\d\d)\]')
message_pattern = re.compile('\"(.+)\"')
# Lambda execution starts here
def handler(event, context):
for record in event['Records']:
# Get the bucket name and key for the new file
bucket = record['s3']['bucket']['name']
key = record['s3']['object']['key']
# Get, read, and split the file into lines
obj = s3.get_object(Bucket=bucket, Key=key)
body = obj['Body'].read()
lines = body.splitlines()
# Match the regular expressions to each line and index the JSON
for line in lines:
line = line.decode("utf-8")
ip = ip_pattern.search(line).group(1)
timestamp = time_pattern.search(line).group(1)
message = message_pattern.search(line).group(1)
document = { "ip": ip, "timestamp": timestamp, "message": message }
r = requests.post(url, auth=awsauth, json=document, headers=headers)
```
Edite las variables para `region` y `host`.
1. [Instale pip](https://pip.pypa.io/en/stable/installation/), si todavía no lo hizo, luego instale las dependencias en un directorio `package` nuevo:
```
cd s3-to-opensearch
pip install --target ./package requests
pip install --target ./package requests_aws4auth
```
Todos los entornos de ejecución de Lambda tienen instalado [Boto3](https://aws.amazon.com/sdk-for-python/), por lo que no es necesario incluirlo en el paquete de implementación.
1. Cree el paquete con el código de la aplicación y las dependencias:
```
cd package
zip -r ../lambda.zip .
cd ..
zip -g lambda.zip sample.py
```
## Crear la función de Lambda
Después de crear el paquete de implementación, puede crear la función de Lambda. Al crear una función, elija un nombre, un tiempo de ejecución (por ejemplo, Python 3.8) y un rol de IAM. El rol de IAM define los permisos para la función. Para obtener instrucciones detalladas, consulte [Create a Lambda function with the consol](https://docs.aws.amazon.com/lambda/latest/dg/get-started-create-function.html) en la *Guía para desarrolladores de AWS Lambda *.
En este ejemplo, se supone que utiliza la consola. Elija Python 3.9 y un rol que tenga permisos de lectura en S3 y permisos de escritura en el OpenSearch servicio, como se muestra en la siguiente captura de pantalla:
![\[Ejemplo de configuración de una función de Lambda\]](http://docs.aws.amazon.com/es_es/opensearch-service/latest/developerguide/images/lambda-function.png)
Después de crear la función, debe agregar un desencadenador. En este ejemplo, queremos que el código se ejecute cada vez que llega un archivo de registros a un bucket de S3:
1. Seleccione **Agregar desencadenador** y Seleccione **S3**.
1. Seleccione el bucket.
1. Para el **Tipo de evento**, elija **PUT**.
1. Para el **Prefijo**, escriba `logs/`.
1. Para el **Sufijo**, escriba `.log`.
1. Reconozca la advertencia de invocación recursiva y elija **Agregar**.
Por último, puede cargar el paquete de implementación:
1. Seleccione **Cargar desde** y **archivo .zip**, luego siga las instrucciones para cargar el paquete de implementación.
1. Una vez finalizada la carga, edite la **Configuración de tiempo de ejecución** y cambie el **Controlador** a `sample.handler`. Esta configuración indica a Lambda el archivo (`sample.py`) y el método (`handler`) que debe ejecutar cuando se produzca un desencadenador.
En este punto, dispone de un conjunto completo de recursos: un depósito para los archivos de registro, una función que se ejecuta cada vez que se añade un archivo de registro al depósito, un código que realiza el análisis y la indexación y un dominio de OpenSearch servicio para la búsqueda y la visualización.
## Prueba de la función de Lambda
Después de crear la función, puede probarla mediante la carga de un archivo en el bucket de Amazon S3. Cree un archivo denominado `sample.log` que contenga los siguientes ejemplos de líneas de registro:
```
12.345.678.90 - [10/Oct/2000:13:55:36 -0700] "PUT /some-file.jpg"
12.345.678.91 - [10/Oct/2000:14:56:14 -0700] "GET /some-file.jpg"
```
Cargue el archivo en la carpeta `logs` del bucket de S3. Para obtener instrucciones, consulte [Carga de un objeto en el bucket](https://docs.aws.amazon.com/AmazonS3/latest/userguide/PuttingAnObjectInABucket.html) en la *Guía del usuario de Amazon Simple Storage Service*.
A continuación, utilice la consola OpenSearch de servicio o los OpenSearch paneles de control para comprobar que el `lambda-s3-index` índice contiene dos documentos. También puede realizar una petición de búsqueda estándar:
```
GET https://domain-name/lambda-s3-index/_search?pretty
{
"hits" : {
"total" : 2,
"max_score" : 1.0,
"hits" : [
{
"_index" : "lambda-s3-index",
"_type" : "_doc",
"_id" : "vTYXaWIBJWV_TTkEuSDg",
"_score" : 1.0,
"_source" : {
"ip" : "12.345.678.91",
"message" : "GET /some-file.jpg",
"timestamp" : "10/Oct/2000:14:56:14 -0700"
}
},
{
"_index" : "lambda-s3-index",
"_type" : "_doc",
"_id" : "vjYmaWIBJWV_TTkEuCAB",
"_score" : 1.0,
"_source" : {
"ip" : "12.345.678.90",
"message" : "PUT /some-file.jpg",
"timestamp" : "10/Oct/2000:13:55:36 -0700"
}
}
]
}
}
```
# Cargar datos de streaming desde Amazon Kinesis Data Streams
Puede cargar datos de streaming desde Kinesis Data Streams OpenSearch a Service. Cuando llegan datos nuevos al flujo de datos, activan una notificación de eventos en Lambda que, a su vez, ejecuta el código personalizado para realizar la indexación. En esta sección, se incluye un código de muestra simple de Python.
## Requisitos previos
Antes de continuar, debe contar con los siguientes recursos.
| Requisito previo | Description (Descripción) |
| --- | --- |
| Amazon Kinesis Data Streams | Fuente de eventos de la función Lambda Para más información, consulte [Kinesis Data Streams](https://docs.aws.amazon.com/kinesis/latest/dev/amazon-kinesis-streams.html). |
| OpenSearch Dominio de servicio | Es el destino de los datos después de que la función Lambda los procesa. Para más información, consulte [Creación de dominios OpenSearch de servicio](createupdatedomains.md#createdomains). |
| Rol de IAM | Esta función debe tener permisos básicos OpenSearch de Servicio, Kinesis y Lambda, como los siguientes: JSON
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"es:ESHttpPost",
"es:ESHttpPut",
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents",
"kinesis:GetShardIterator",
"kinesis:GetRecords",
"kinesis:DescribeStream",
"kinesis:ListStreams"
],
"Resource": "*"
}
]
}
``` El rol debe tener la siguiente relación de confianza: JSON
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "lambda.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}
``` Para más información, consulte [Creación de roles de IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create.html) en la *Guía del usuario de IAM*. |
# Crear la función de Lambda
Siga las instrucciones de [Crear el paquete de implementación de Lambda](integrations-s3-lambda.md#integrations-s3-lambda-deployment-package), pero cree un directorio denominado `kinesis-to-opensearch` y utilice el siguiente código para `sample.py`:
```
import base64
import boto3
import json
import requests
from requests_aws4auth import AWS4Auth
region = '' # e.g. us-west-1
service = 'es'
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)
host = '' # the OpenSearch Service domain, e.g. https://search-mydomain.us-west-1.es.amazonaws.com
index = 'lambda-kine-index'
datatype = '_doc'
url = host + '/' + index + '/' + datatype + '/'
headers = { "Content-Type": "application/json" }
def handler(event, context):
count = 0
for record in event['Records']:
id = record['eventID']
timestamp = record['kinesis']['approximateArrivalTimestamp']
# Kinesis data is base64-encoded, so decode here
message = base64.b64decode(record['kinesis']['data'])
# Create the JSON document
document = { "id": id, "timestamp": timestamp, "message": message }
# Index the document
r = requests.put(url + id, auth=awsauth, json=document, headers=headers)
count += 1
return 'Processed ' + str(count) + ' items.'
```
Edite las variables para `region` y `host`.
[Instale pip](https://pip.pypa.io/en/stable/installation/), si todavía no lo hizo, luego utilice los siguientes comandos para instalar las dependencias:
```
cd kinesis-to-opensearch
pip install --target ./package requests
pip install --target ./package requests_aws4auth
```
A continuación, siga las instrucciones de [Crear la función de Lambda](integrations-s3-lambda.md#integrations-s3-lambda-create), pero especifique el rol de IAM de [Requisitos previos](integrations-kinesis.md#integrations-kinesis-lambda-prereq) y la configuración siguiente para el desencadenador:
+ **Flujo de Kinesis**: su flujo de Kinesis
+ **Tamaño del lote**: 100
+ **Posición inicial**: Trim horizon
Para más información, consulte [¿Qué es Amazon Kinesis Data Streams?](https://docs.aws.amazon.com/streams/latest/dev/working-with-kinesis.html) en la *Guía para desarrolladores de Amazon Kinesis Data Streams*.
En este punto, dispone de un conjunto completo de recursos: una transmisión de datos de Kinesis, una función que se ejecuta después de que la transmisión reciba nuevos datos y los indexe, y un dominio de OpenSearch servicio para la búsqueda y la visualización.
# Prueba de la función de Lambda
Después de crear la función, puede probarla al agregar un registro al flujo de datos mediante la AWS CLI:
```
aws kinesis put-record --stream-name test --data "My test data." --partition-key partitionKey1 --region us-west-1
```
A continuación, utilice la consola OpenSearch de servicio o los OpenSearch paneles de control para comprobar que `lambda-kine-index` contiene un documento. También puede utilizar la solicitud siguiente:
```
GET https://domain-name/lambda-kine-index/_search
{
"hits" : [
{
"_index": "lambda-kine-index",
"_type": "_doc",
"_id": "shardId-000000000000:49583511615762699495012960821421456686529436680496087042",
"_score": 1,
"_source": {
"timestamp": 1523648740.051,
"message": "My test data.",
"id": "shardId-000000000000:49583511615762699495012960821421456686529436680496087042"
}
}
]
}
```
# Carga de datos de streaming desde Amazon DynamoDB
Puede utilizarlos AWS Lambda para enviar datos a su dominio de OpenSearch servicio desde Amazon DynamoDB. Cuando llegan datos nuevos a la tabla de base de datos, activan una notificación de eventos en Lambda que, a su vez, ejecuta el código personalizado para realizar la indexación.
## Requisitos previos
Antes de continuar, debe contar con los siguientes recursos.
| Requisito previo | Description (Descripción) |
| --- | --- |
| Tabla de DynamoDB | La tabla contiene los datos de origen. Para más información, consulte [Operaciones básicas en tablas de DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/WorkingWithTables.Basics.html) en la *Guía para desarrolladores de Amazon DynamoDB*.La tabla debe residir en la misma región que su dominio de OpenSearch servicio y tener una transmisión configurada como **Nueva** imagen. Para más información, consulte [Habilitación de un flujo](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html#Streams.Enabling). |
| OpenSearch Dominio de servicio | Es el destino de los datos después de que la función de Lambda los procesa. Para más información, consulte [Creación de dominios OpenSearch de servicio](createupdatedomains.md#createdomains). |
| rol de IAM | Esta función debe tener permisos básicos OpenSearch de ejecución de Servicios, DynamoDB y Lambda, como los siguientes: JSON
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"es:ESHttpPost",
"es:ESHttpPut",
"dynamodb:DescribeStream",
"dynamodb:GetRecords",
"dynamodb:GetShardIterator",
"dynamodb:ListStreams",
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents"
],
"Resource": "*"
}
]
}
``` El rol debe tener la siguiente relación de confianza: JSON
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "lambda.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}
``` Para más información, consulte [Creación de roles de IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create.html) en la *Guía del usuario de IAM*. |
## Crear la función de Lambda
Siga las instrucciones de [Crear el paquete de implementación de Lambda](integrations-s3-lambda.md#integrations-s3-lambda-deployment-package), pero cree un directorio denominado `ddb-to-opensearch` y utilice el siguiente código para `sample.py`:
```
import boto3
import requests
from requests_aws4auth import AWS4Auth
region = '' # e.g. us-east-1
service = 'es'
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)
host = '' # the OpenSearch Service domain, e.g. https://search-mydomain.us-west-1.es.amazonaws.com
index = 'lambda-index'
datatype = '_doc'
url = host + '/' + index + '/' + datatype + '/'
headers = { "Content-Type": "application/json" }
def handler(event, context):
count = 0
for record in event['Records']:
# Get the primary key for use as the OpenSearch ID
id = record['dynamodb']['Keys']['id']['S']
if record['eventName'] == 'REMOVE':
r = requests.delete(url + id, auth=awsauth)
else:
document = record['dynamodb']['NewImage']
r = requests.put(url + id, auth=awsauth, json=document, headers=headers)
count += 1
return str(count) + ' records processed.'
```
Edite las variables para `region` y `host`.
[Instale pip](https://pip.pypa.io/en/stable/installation/), si todavía no lo hizo, luego utilice los siguientes comandos para instalar las dependencias:
```
cd ddb-to-opensearch
pip install --target ./package requests
pip install --target ./package requests_aws4auth
```
A continuación, siga las instrucciones de [Crear la función de Lambda](integrations-s3-lambda.md#integrations-s3-lambda-create), pero especifique el rol de IAM de [Requisitos previos](#integrations-dynamodb-prereq) y la configuración siguiente para el desencadenador:
+ **Tabla**: tabla de DynamoDB
+ **Tamaño del lote**: 100
+ **Posición inicial**: Trim horizon
Para más información, consulte [Process New Items with DynamoDB Streams and Lambda](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.Lambda.Tutorial.html) en la *Guía para desarrolladores de Amazon DynamoDB*.
En este punto, dispone de un conjunto completo de recursos: una tabla de DynamoDB para los datos de origen, un flujo de cambios de DynamoDB en la tabla, una función que se ejecuta después de que los datos de origen cambien e indexa esos cambios, y un dominio de servicio para la búsqueda y la visualización. OpenSearch
## Prueba de la función de Lambda
Después de crear la función, puede probarla al agregar un elemento a la tabla de DynamoDB mediante la AWS CLI:
```
aws dynamodb put-item --table-name test --item '{"director": {"S": "Kevin Costner"},"id": {"S": "00001"},"title": {"S": "The Postman"}}' --region us-west-1
```
A continuación, utilice la consola de OpenSearch servicio o los OpenSearch paneles de control para comprobar que contiene un documento. `lambda-index` También puede utilizar la solicitud siguiente:
```
GET https://domain-name/lambda-index/_doc/00001
{
"_index": "lambda-index",
"_type": "_doc",
"_id": "00001",
"_version": 1,
"found": true,
"_source": {
"director": {
"S": "Kevin Costner"
},
"id": {
"S": "00001"
},
"title": {
"S": "The Postman"
}
}
}
```
# Carga de datos de streaming desde Amazon Data Firehose
Firehose admite el OpenSearch servicio como destino de entrega. Para obtener instrucciones sobre cómo cargar datos de streaming en el OpenSearch servicio, consulte [Crear una transmisión de entrega de Kinesis Data Firehose OpenSearch ](https://docs.aws.amazon.com/firehose/latest/dev/basic-create.html) [y elegir un servicio para su destino en la Guía para](https://docs.aws.amazon.com/firehose/latest/dev/create-destination.html#create-destination-elasticsearch) *desarrolladores de Amazon Data Firehose*.
Antes de cargar datos en OpenSearch Service, es posible que deba realizar transformaciones en los datos. Para obtener más información acerca de cómo utilizar las funciones de Lambda para realizar esta tarea, consulte [Amazon Kinesis Data Firehose Data Transformation](https://docs.aws.amazon.com/firehose/latest/dev/data-transformation.html) en la misma guía.
Al configurar una transmisión de entrega, Firehose incluye una función de IAM con un solo clic que le proporciona el acceso a los recursos que necesita para enviar datos al OpenSearch Servicio, hacer copias de seguridad de los datos en Amazon S3 y transformar los datos con Lambda. Debido a la complejidad que supone la creación de ese rol manualmente, recomendamos utilizar el rol proporcionado.
# Carga de datos de streaming desde Amazon CloudWatch
Puedes cargar datos de streaming desde CloudWatch Logs a tu dominio OpenSearch de Service mediante una suscripción a CloudWatch Logs. Para obtener información sobre CloudWatch las suscripciones de Amazon, consulta [Procesamiento en tiempo real de datos de registro con suscripciones](https://docs.aws.amazon.com/AmazonCloudWatch/latest/DeveloperGuide/Subscriptions.html). Para obtener información sobre la configuración, consulte [Transmisión de datos de CloudWatch registros a Amazon OpenSearch Service](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/CWL_OpenSearch_Stream.html) en la *Guía para CloudWatch desarrolladores de Amazon*.
## Cargando datos de streaming desde AWS IoT
Puede enviar datos AWS IoT mediante [reglas](https://docs.aws.amazon.com/iot/latest/developerguide/iot-rules.html). Para obtener más información, consulta la [OpenSearch](https://docs.aws.amazon.com/iot/latest/developerguide/opensearch-rule-action.html)acción en la *Guía para AWS IoT desarrolladores*.