Ingestion de données dans des collections Amazon OpenSearch Serverless - Amazon OpenSearch Service
Autorisations minimales requisesOpenSearch IngestionFluent BitAmazon Data FirehoseFluentdGoJavaJavaScriptLogstashPythonRubyAutres clients

Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Ingestion de données dans des collections Amazon OpenSearch Serverless

Ces sections fournissent des informations sur les pipelines d'ingestion pris en charge pour l'ingestion de données dans les collections Amazon OpenSearch Serverless. Ils couvrent également certains des clients que vous pouvez utiliser pour interagir avec les opérations de l' OpenSearch API. Vos clients doivent être compatibles avec la version OpenSearch 2.x afin de pouvoir s'intégrer à OpenSearch Serverless.

Autorisations minimales requises

Afin d'ingérer des données dans une collection OpenSearch sans serveur, le principal qui écrit les données doit disposer des autorisations minimales suivantes attribuées dans une politique d'accès aux données :

[
   {
      "Rules":[
         {
            "ResourceType":"index",
            "Resource":[
               "index/target-collection/logs"
            ],
            "Permission":[
               "aoss:CreateIndex",
               "aoss:WriteDocument",
               "aoss:UpdateIndex"
            ]
         }
      ],
      "Principal":[
         "arn:aws:iam::123456789012:user/my-user"
      ]
   }
]

Les autorisations peuvent être plus étendues si vous prévoyez d'écrire dans des index supplémentaires. Par exemple, plutôt que de spécifier un seul index cible, vous pouvez autoriser l'accès à tous les index (index/target-collection/*) ou à un sous-ensemble d'index (index/target-collection/logs*).

Pour une référence de toutes les opérations OpenSearch d'API disponibles et de leurs autorisations associées, consultezOpérations et plug-ins pris en charge dans Amazon OpenSearch Serverless.

OpenSearch Ingestion

Plutôt que d'utiliser un client tiers pour envoyer des données directement à une collection OpenSearch sans serveur, vous pouvez utiliser Amazon OpenSearch Ingestion. Vous configurez vos producteurs de données pour qu'ils envoient des données à OpenSearch Ingestion, qui les fournit automatiquement à la collection que vous spécifiez. Vous pouvez également configurer OpenSearch Ingestion pour transformer vos données avant de les livrer. Pour plus d’informations, consultez OpenSearch Ingestion d'Amazon.

Un pipeline d' OpenSearch ingestion a besoin d'une autorisation pour écrire dans une collection OpenSearch sans serveur configurée comme récepteur. Ces autorisations incluent la possibilité de décrire la collection et de lui envoyer des requêtes HTTP. Pour obtenir des instructions sur OpenSearch l'utilisation d'Ingestion pour ajouter des données à une collection, reportez-vous àAutoriser les pipelines OpenSearch Amazon Ingestion à accéder aux collections.

Pour commencer à utiliser OpenSearch Ingestion, voirTutoriel : Ingestion de données dans une collection à l'aide d'Amazon OpenSearch Ingestion.

Fluent Bit

Vous pouvez utiliser l'AWS image Fluent Bit et le plugin OpenSearch de sortie pour ingérer des données dans des collections OpenSearch sans serveur.

Note

Vous devez disposer de la version 2.30.0 ou ultérieure de l'image AWS for Fluent Bit pour pouvoir intégrer Serverless. OpenSearch

Exemple de configuration :

Cet exemple de section de sortie du fichier de configuration montre comment utiliser une collection OpenSearch Serverless comme destination. L'ajout important est le paramètre AWS_Service_Name, qui est aoss. Host est le point de terminaison de la collection.

[OUTPUT]
    Name  opensearch
    Match *
    Host  collection-endpoint.us-west-2.aoss.amazonaws.com
    Port  443
    Index  my_index
    Trace_Error On
    Trace_Output On
    AWS_Auth On
    AWS_Region <region>
    AWS_Service_Name aoss
    tls     On
    Suppress_Type_Name On

Amazon Data Firehose

Firehose prend en charge le mode OpenSearch Serverless comme destination de livraison. Pour obtenir des instructions sur l'envoi de données vers OpenSearch Serverless, consultez Creating a Kinesis Data Firehose Delivery Stream et OpenSearch Choose Serverless for Your Destination dans le manuel Amazon Data Firehose Developer Guide.

Le rôle IAM que vous fournissez à Firehose pour la livraison doit être spécifié dans une politique d'accès aux données avec aoss:WriteDocument l'autorisation minimale pour la collecte cible, et vous devez disposer d'un index préexistant auquel envoyer des données. Pour plus d’informations, consultez Autorisations minimales requises.

Avant d'envoyer des données vers OpenSearch Serverless, vous devrez peut-être effectuer des transformations sur les données. Pour en savoir plus sur l'utilisation des fonctions Lambda pour effectuer cette tâche, consultez Transformation de données Amazon Kinesis Data Firehose dans le même guide.

Fluentd

Vous pouvez utiliser le OpenSearch plugin Fluentd pour collecter des données à partir de votre infrastructure, de vos conteneurs et de vos appareils réseau et les envoyer vers des collections OpenSearch sans serveur. Calyptia gère une distribution de Fluentd qui contient toutes les dépendances en aval de Ruby et SSL.

Pour utiliser Fluentd pour envoyer des données vers Serverless OpenSearch

  1. Téléchargez la version 1.4.2 ou ultérieure de Calyptia Fluentd depuis https://www.fluentd.org/download. Cette version inclut le OpenSearch plugin par défaut, qui supporte OpenSearch Serverless.

  2. Installez le package . Suivez les instructions de la documentation Fluentd en fonction de votre système d'exploitation :

  3. Ajoutez une configuration qui envoie des données à OpenSearch Serverless. Cet exemple de configuration envoie le message « test » à une seule collection. Vérifiez les points suivants :

    • Pourhost, spécifiez le point de terminaison de votre collection OpenSearch Serverless.

    • Pour aws_service_name, spécifiez aoss.

    <source>
@type sample
tag test
test {"hello":"world"}
</source>

<match test>
@type opensearch
host https://collection-endpoint.us-east-1.aoss.amazonaws.com
port 443
index_name fluentd
aws_service_name aoss
</match>

  4. Exécutez Calyptia Fluentd pour commencer à envoyer des données à la collection. Par exemple, sur Mac, vous pouvez exécuter la commande suivante :

    sudo launchctl load /Library/LaunchDaemons/calyptia-fluentd.plist

Go

L'exemple de code suivant utilise le client opensearch-go pour Go afin d'établir une connexion sécurisée avec la collection OpenSearch Serverless spécifiée et de créer un index unique. Vous devez fournir des valeurs pour region et host.

package main

import (
  "context"
  "log"
  "strings"
  "github.com/aws/aws-sdk-go-v2/aws"
  "github.com/aws/aws-sdk-go-v2/config"
  opensearch "github.com/opensearch-project/opensearch-go/v2"
  opensearchapi "github.com/opensearch-project/opensearch-go/v2/opensearchapi"
  requestsigner "github.com/opensearch-project/opensearch-go/v2/signer/awsv2"
)

const endpoint = "" // serverless collection endpoint

func main() {
	ctx := context.Background()

	awsCfg, err := config.LoadDefaultConfig(ctx,
		config.WithRegion("<AWS_REGION>"),
		config.WithCredentialsProvider(
			getCredentialProvider("<AWS_ACCESS_KEY>", "<AWS_SECRET_ACCESS_KEY>", "<AWS_SESSION_TOKEN>"),
		),
	)
	if err != nil {
		log.Fatal(err) // don't log.fatal in a production-ready app
	}

	// create an AWS request Signer and load AWS configuration using default config folder or env vars.
	signer, err := requestsigner.NewSignerWithService(awsCfg, "aoss") // "aoss" for Amazon OpenSearch Serverless
	if err != nil {
		log.Fatal(err) // don't log.fatal in a production-ready app
	}

	// create an opensearch client and use the request-signer
	client, err := opensearch.NewClient(opensearch.Config{
		Addresses: []string{endpoint},
		Signer:    signer,
	})
	if err != nil {
		log.Fatal("client creation err", err)
	}

	indexName := "go-test-index"

  // define index mapping
	mapping := strings.NewReader(`{
	 "settings": {
	   "index": {
	        "number_of_shards": 4
	        }
	      }
	 }`)

	// create an index
	createIndex := opensearchapi.IndicesCreateRequest{
		Index: indexName,
    Body: mapping,
	}
	createIndexResponse, err := createIndex.Do(context.Background(), client)
	if err != nil {
		log.Println("Error ", err.Error())
		log.Println("failed to create index ", err)
		log.Fatal("create response body read err", err)
	}
	log.Println(createIndexResponse)

	// delete the index
	deleteIndex := opensearchapi.IndicesDeleteRequest{
		Index: []string{indexName},
	}

	deleteIndexResponse, err := deleteIndex.Do(context.Background(), client)
	if err != nil {
		log.Println("failed to delete index ", err)
		log.Fatal("delete index response body read err", err)
	}
	log.Println("deleting index", deleteIndexResponse)
}

func getCredentialProvider(accessKey, secretAccessKey, token string) aws.CredentialsProviderFunc {
	return func(ctx context.Context) (aws.Credentials, error) {
		c := &aws.Credentials{
			AccessKeyID:     accessKey,
			SecretAccessKey: secretAccessKey,
			SessionToken:    token,
		}
		return *c, nil
	}
}

Java

L'exemple de code suivant utilise le client opensearch-java pour Java afin d'établir une connexion sécurisée avec la collection OpenSearch Serverless spécifiée et de créer un index unique. Vous devez fournir des valeurs pour region et host.

La différence importante par rapport aux domaines OpenSearch de service réside dans le nom du service (aossau lieu dees).

// import OpenSearchClient to establish connection to OpenSearch Serverless collection
import org.opensearch.client.opensearch.OpenSearchClient;

SdkHttpClient httpClient = ApacheHttpClient.builder().build();
// create an opensearch client and use the request-signer
OpenSearchClient client = new OpenSearchClient(
    new AwsSdk2Transport(
        httpClient,
        "...us-west-2.aoss.amazonaws.com", // serverless collection endpoint
        "aoss" // signing service name
        Region.US_WEST_2, // signing service region
        AwsSdk2TransportOptions.builder().build()
    )
);

String index = "sample-index";

// create an index
CreateIndexRequest createIndexRequest = new CreateIndexRequest.Builder().index(index).build();
CreateIndexResponse createIndexResponse = client.indices().create(createIndexRequest);
System.out.println("Create index reponse: " + createIndexResponse);

// delete the index
DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest.Builder().index(index).build();
DeleteIndexResponse deleteIndexResponse = client.indices().delete(deleteIndexRequest);
System.out.println("Delete index reponse: " + deleteIndexResponse);

httpClient.close();

L'exemple de code suivant établit à nouveau une connexion sécurisée, puis recherche un index.

import org.opensearch.client.opensearch.OpenSearchClient;

SdkHttpClient httpClient = ApacheHttpClient.builder().build();

OpenSearchClient client = new OpenSearchClient(
    new AwsSdk2Transport(
        httpClient,
        "...us-west-2.aoss.amazonaws.com", // serverless collection endpoint
        "aoss" // signing service name
        Region.US_WEST_2, // signing service region
        AwsSdk2TransportOptions.builder().build()
    )
);

Response response = client.generic()
    .execute(
        Requests.builder()
            .endpoint("/" + "users" + "/_search?typed_keys=true")
            .method("GET")
            .json("{"
                + "    \"query\": {"
                + "        \"match_all\": {}"
                + "    }"
                + "}")
            .build());

httpClient.close();

JavaScript

L'exemple de code suivant utilise le client opensearch-js JavaScript pour établir une connexion sécurisée avec la collection OpenSearch Serverless spécifiée, créer un index unique, ajouter un document et supprimer l'index. Vous devez fournir des valeurs pour node et region.

La différence importante par rapport aux domaines OpenSearch de service réside dans le nom du service (aossau lieu dees).

Version 3

Cet exemple utilise la version 3 du SDK pour JavaScript le fichier Node.js.

const { defaultProvider } = require('@aws-sdk/credential-provider-node');
const { Client } = require('@opensearch-project/opensearch');
const { AwsSigv4Signer } = require('@opensearch-project/opensearch/aws');

async function main() {
    // create an opensearch client and use the request-signer
    const client = new Client({
        ...AwsSigv4Signer({
            region: 'us-west-2',
            service: 'aoss',
            getCredentials: () => {
                const credentialsProvider = defaultProvider();
                return credentialsProvider();
            },
        }),
        node: '' # // serverless collection endpoint
    });

    const index = 'movies';

    // create index if it doesn't already exist
    if (!(await client.indices.exists({ index })).body) {
        console.log((await client.indices.create({ index })).body);
    }

    // add a document to the index
    const document = { foo: 'bar' };
    const response = await client.index({
        id: '1',
        index: index,
        body: document,
    });
    console.log(response.body);

    // delete the index
    console.log((await client.indices.delete({ index })).body);
}

main();
Version 2

Cet exemple utilise la version 2 du SDK pour JavaScript le fichier Node.js.

const AWS = require('aws-sdk');
const { Client } = require('@opensearch-project/opensearch');
const { AwsSigv4Signer } = require('@opensearch-project/opensearch/aws');

async function main() {
    // create an opensearch client and use the request-signer
    const client = new Client({
        ...AwsSigv4Signer({
            region: 'us-west-2',
            service: 'aoss',
            getCredentials: () =>
                new Promise((resolve, reject) => {
                    AWS.config.getCredentials((err, credentials) => {
                        if (err) {
                            reject(err);
                        } else {
                            resolve(credentials);
                        }
                    });
                }),
        }),
        node: '' # // serverless collection endpoint
    });

    const index = 'movies';

    // create index if it doesn't already exist
    if (!(await client.indices.exists({ index })).body) {
        console.log((await client.indices.create({
            index
        })).body);
    }

    // add a document to the index
    const document = {
        foo: 'bar'
    };
    const response = await client.index({
        id: '1',
        index: index,
        body: document,
    });
    console.log(response.body);

    // delete the index
    console.log((await client.indices.delete({ index })).body);
}

main();

Logstash

Vous pouvez utiliser le OpenSearch plugin Logstash pour publier des journaux dans des collections OpenSearch Serverless.

Pour utiliser Logstash pour envoyer des données vers Serverless OpenSearch

  1. Installez la version 2.0.0 ou ultérieure du logstash-output-opensearchplugin à l'aide de Docker ou Linux.

    Docker

    Docker héberge le logiciel Logstash OSS avec le plugin de OpenSearch sortie préinstallé : opensearchproject/ -output-plugin. logstash-oss-with-opensearch Vous pouvez extraire l'image comme n'importe quelle autre image :

    docker pull opensearchproject/logstash-oss-with-opensearch-output-plugin:latest
    Linux

    Si vous ne l'avez pas déjà fait, installez la dernière version de Logstash. Ensuite, installez la version 2.0.0 du plugin de sortie :

    cd logstash-8.5.0/
bin/logstash-plugin install --version 2.0.0 logstash-output-opensearch

    Si le plugin est déjà installé, mettez-le à jour vers la dernière version :

    bin/logstash-plugin update logstash-output-opensearch

    À partir de la version 2.0.0 du plugin, le AWS SDK utilise la version 3. Si vous utilisez une version de Logstash antérieure à la version 8.4.0, vous devez supprimer tous les AWS plugins préinstallés et installer le plugin : logstash-integration-aws

    /usr/share/logstash/bin/logstash-plugin remove logstash-input-s3
/usr/share/logstash/bin/logstash-plugin remove logstash-input-sqs
/usr/share/logstash/bin/logstash-plugin remove logstash-output-s3
/usr/share/logstash/bin/logstash-plugin remove logstash-output-sns
/usr/share/logstash/bin/logstash-plugin remove logstash-output-sqs
/usr/share/logstash/bin/logstash-plugin remove logstash-output-cloudwatch

/usr/share/logstash/bin/logstash-plugin install --version 0.1.0.pre logstash-integration-aws

  2. Pour que le plugin OpenSearch de sortie fonctionne avec OpenSearch Serverless, vous devez apporter les modifications suivantes à la section de opensearch sortie de logstash.conf :

    • Spécifiez aoss comme service_name sous auth_type.

    • Spécifiez le point de terminaison de votre collection pour hosts.

    • Ajoutez les paramètres default_server_major_version et legacy_template. Ces paramètres sont nécessaires pour que le plugin fonctionne avec OpenSearch Serverless.

    output {
  opensearch {
    hosts => "collection-endpoint:443"
    auth_type => {
      ...
      service_name => 'aoss'
    }
    default_server_major_version => 2
    legacy_template => false
  }
}

    Cet exemple de fichier de configuration prend ses entrées à partir des fichiers d'un compartiment S3 et les envoie à une collection OpenSearch Serverless :

    input {
  s3  {
    bucket => "my-s3-bucket"
    region => "us-east-1"
  }
}

output {
  opensearch {
    ecs_compatibility => disabled
    hosts => "https://my-collection-endpoint.us-east-1.aoss.amazonaws.com:443"
    index => my-index
    auth_type => {
      type => 'aws_iam'
      aws_access_key_id => 'your-access-key'
      aws_secret_access_key => 'your-secret-key'
      region => 'us-east-1'
      service_name => 'aoss'
    }
    default_server_major_version => 2
    legacy_template => false
  }
}

  3. Ensuite, lancez Logstash avec la nouvelle configuration pour tester le plugin :

    bin/logstash -f config/test-plugin.conf

Python

L'exemple de code suivant utilise le client opensearch-py pour Python afin d'établir une connexion sécurisée avec la collection OpenSearch Serverless spécifiée, de créer un index unique et d'effectuer une recherche dans cet index. Vous devez fournir des valeurs pour region et host.

La différence importante par rapport aux domaines OpenSearch de service réside dans le nom du service (aossau lieu dees).

from opensearchpy import OpenSearch, RequestsHttpConnection, AWSV4SignerAuth
import boto3

host = ''  # serverless collection endpoint, without https://
region = ''  # e.g. us-east-1

service = 'aoss'
credentials = boto3.Session().get_credentials()
auth = AWSV4SignerAuth(credentials, region, service)

# create an opensearch client and use the request-signer
client = OpenSearch(
    hosts=[{'host': host, 'port': 443}],
    http_auth=auth,
    use_ssl=True,
    verify_certs=True,
    connection_class=RequestsHttpConnection,
    pool_maxsize=20,
)

# create an index
index_name = 'books-index'
create_response = client.indices.create(
    index_name
)

print('\nCreating index:')
print(create_response)

# index a document
document = {
  'title': 'The Green Mile',
  'director': 'Stephen King',
  'year': '1996'
}

response = client.index(
    index = 'books-index',
    body = document,
    id = '1'
)


# delete the index
delete_response = client.indices.delete(
    index_name
)

print('\nDeleting index:')
print(delete_response)

Ruby

La opensearch-aws-sigv4 gemme fournit un accès à OpenSearch Serverless, ainsi qu'à OpenSearch Service, prêt à l'emploi. Elle possède toutes les fonctions du client opensearch-ruby, car elle est une dépendance de cette gemme.

Lors de l'instanciation du signataire Sigv4, spécifiez aoss comme nom de service :

require 'opensearch-aws-sigv4'
require 'aws-sigv4'

signer = Aws::Sigv4::Signer.new(service: 'aoss',
                                region: 'us-west-2',
                                access_key_id: 'key_id',
                                secret_access_key: 'secret')

# create an opensearch client and use the request-signer
client = OpenSearch::Aws::Sigv4Client.new(
  { host: 'https://your.amz-opensearch-serverless.endpoint',
    log: true },
  signer)

# create an index
index = 'prime'
client.indices.create(index: index)

# insert data
client.index(index: index, id: '1', body: { name: 'Amazon Echo', 
                                            msrp: '5999', 
                                            year: 2011 })

# query the index
client.search(body: { query: { match: { name: 'Echo' } } })

# delete index entry
client.delete(index: index, id: '1')

# delete the index
client.indices.delete(index: index)

Signature des demandes HTTP avec d'autres clients

Les exigences suivantes s'appliquent lors de la signature de demandes destinées à des collections OpenSearch sans serveur lorsque vous créez des requêtes HTTP avec d'autres clients.

  • Vous devez spécifier le nom du service sous la forme aoss.

  • L'en-tête x-amz-content-sha256 est obligatoire pour toutes les requêtes AWS  Signature Version 4. Il fournit un hachage de la charge utile de la requête. S'il existe une charge utile de requête, définissez la valeur sur son hachage de chiffrement SHA (Secure Hash Algorithm, algorithme de hachage sécurisé) (SHA256). S'il n'existe aucune charge utile de requête, définissez la valeur sur e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855, qui est le hachage d'une chaîne vide.

Indexation avec cURL

L'exemple de demande suivant utilise la bibliothèque de demandes d'URL du client (cURL) pour envoyer un seul document vers un index nommé movies-index dans une collection :

curl -XPOST \
    --user "$AWS_ACCESS_KEY_ID":"$AWS_SECRET_ACCESS_KEY" \
    --aws-sigv4 "aws:amz:us-east-1:aoss" \
    --header "x-amz-content-sha256: $REQUEST_PAYLOAD_SHA_HASH" \
    --header "x-amz-security-token: $AWS_SESSION_TOKEN" \
    "https://my-collection-endpoint.us-east-1.aoss.amazonaws.com/movies-index/_doc" \
    -H "Content-Type: application/json" -d '{"title": "Shawshank Redemption"}'

Indexation avec Postman

L'image suivante montre comment envoyer une demande à une collection à l'aide de Postman. Pour obtenir des instructions d'authentification, consultez le flux de travail d'authentification Authentifier avec AWS signature dans Postman.