Amazon OpenSearch Serverless コレクションへのデータの取り込み - Amazon OpenSearch Service
必要な最小限のアクセス許可OpenSearch IngestionFluent BitAmazon Data FirehoseGoJavaJavaScriptLogstashPythonRubyその他のクライアント

翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

Amazon OpenSearch Serverless コレクションへのデータの取り込み

これらのセクションでは、Amazon OpenSearch Serverless コレクションへのデータインジェストでサポートされている取り込みパイプラインについて詳しく説明します。OpenSearch API オペレーションの操作に使用できるクライアントの一部についても説明しています。OpenSearch Serverless と統合するには、クライアントが OpenSearch 2.x と互換性がある必要があります。

必要な最小限のアクセス許可

OpenSearch Serverless コレクションにデータを取り込むには、データを書き込むプリンシパルに、データアクセスポリシーで次の最小限のアクセス許可が割り当てられている必要があります。

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

追加のインデックスに書き込む場合は、アクセス許可をより幅広く指定できます。例えば、1 つのターゲットインデックスを指定するのではなく、すべてのインデックス (index/target-collection/*) または一部のインデックス (index/target-collection/logs*) へのアクセス許可を付与することができます。

使用可能なすべての OpenSearch API オペレーションとそれに関連するアクセス許可のリファレンスについては、「Amazon OpenSearch Serverless でサポートされているオペレーションとプラグイン」を参照してください。

OpenSearch Ingestion

OpenSearch サーバーレスコレクションにデータを直接送信する場合、サードパーティのクライアントを使用する代わりに Amazon OpenSearch Ingestion を使用できます。OpenSearch Ingestion にデータを送信するようにデータプロデューサーを設定すると、指定したコレクションにデータが自動的に送信されます。また、OpenSearch Ingestion で、送信前にデータを変換するように設定することもできます。詳細については、「Amazon OpenSearch Ingestion の概要」を参照してください。

シンクとして設定されている OpenSearch Serverless コレクションに書き込みを行うときは、OpenSearch Ingestion パイプラインにアクセス権限が必要になります。これらのアクセス権限には、コレクションを記述しそこに HTTP リクエストを送信できることが含まれます。OpenSearch Ingestion を使用してコレクションにデータを追加する手順については、「Amazon OpenSearch Ingestion パイプラインにコレクションへのアクセス権を付与する」を参照してください。

OpenSearch Ingestion の使用を開始するには、「チュートリアル: Amazon OpenSearch Ingestion を使用してデータをコレクションに取り込む」を参照してください。

Fluent Bit

OpenSearch Serverless コレクションにデータを取り込むために、AWS for Fluent Bit イメージと OpenSearch 出力プラグインを使用できます。

注記

OpenSearch Serverless と統合するには、 AWS for Fluent Bit イメージのバージョン 2.30.0 以降が必要です。

設定例:

設定ファイルのこのサンプル出力セクションは、OpenSearch Serverless コレクションを宛先として使用する方法を示しています。重要な追加項目は AWS_Service_Name パラメータで、これは aoss です。Host はコレクションエンドポイントです。

[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 は配信先として OpenSearch Serverless をサポートしています。OpenSearch Serverless にデータを送信する方法については、「Amazon Data Firehose Developer Guide」の「Creating a Kinesis Data Firehose Delivery Stream」および「Choose OpenSearch Serverless for Your Destination」を参照してください。

配信のために Firehose に提供する IAM ロールは、ターゲットコレクションに対する最小限の aoss:WriteDocument 許可を持つデータアクセスポリシー内で指定する必要があります。また、データを送信する既存のインデックスが必要です。詳細については、「必要な最小限のアクセス許可」を参照してください。

データを OpenSearch Serverless に送信する前に、データの変換が必要になる場合があります。Lambda 関数を使用してこのタスクを実行する方法については、このガイドの「Amazon Kinesis Data Firehose データ変換」を参照してください。

Go

以下のサンプルコードでは、Go 用の opensearch-go クライアントを使用して、指定された OpenSearch Serverless コレクションへのセキュアな接続を確立し、単一のインデックスを作成します。region および 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

以下のサンプルコードでは、Java 用の opensearch-java クライアントを使用して、指定された OpenSearch Serverless コレクションへのセキュアな接続を確立し、単一のインデックスを作成します。region および host の値を指定する必要があります。

OpenSearch Service ドメインとの重要な違いは、サービス名です (es ではなく aoss)。

// 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();

次のサンプルコードは、安全な接続を再度確立し、インデックスを検索します。

import org.opensearch.client.opensearch.OpenSearchClient;
>>>>>>> aoss-slr-update

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

以下のサンプルコードでは、JavaScript 用の opensearch-js クライアントを使用して、指定された OpenSearch Serverless コレクションへのセキュアな接続の確立、単一のインデックスの作成、ドキュメントの追加、およびインデックスの削除を実行します。node および region の値を指定する必要があります。

OpenSearch Service ドメインとの重要な違いは、サービス名です (es ではなく aoss)。

Version 3

この例では SDK for JavaScript in Node.js の バージョン 3 を使用します。

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

この例では SDK for JavaScript in Node.js の バージョン 2 を使用します。

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

OpenSearch Serverless コレクションにログを公開するときは、Logstash OpenSearch プラグインを使用します。

Logstash を使用して OpenSearch Serverless にデータを送信するには

  1. Docker または Linux を使用して logstash-output-opensearch プラグインのバージョン 2.0.0 以降をインストールします。

    Docker

    Docker は、OpenSearch 出力プラグイン (opensearchproject/logstash-oss-with-opensearch-output-plugin) がプリインストールされた Logstash OSS ソフトウェアをホストしています。他のイメージと同じように次のようにイメージをプルできます。

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

    まだ Logstash の最新バージョンをインストールしていない場合は、インストールします。次に、出力プラグインのバージョン 2.0.0 を次のようにインストールします。

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

    プラグインが既にインストールされている場合は、次のように最新バージョンに更新します。

    bin/logstash-plugin update logstash-output-opensearch

    プラグインのバージョン 2.0.0 以降、 AWS SDK はバージョン 3 を使用します。8.4.0 より前のバージョンの Logstash を使用している場合は、プリインストールされている AWS プラグインをすべて削除し、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. OpenSearch 出力プラグインが OpenSearch Serverless で動作するためには、logstash.conf の opensearch 出力セクションに次の変更を加える必要があります。

    • auth_type の service_name として aoss を指定します。

    • hosts のコレクションエンドポイントを指定します。

    • パラメータ default_server_major_version および legacy_template を追加します。これらのパラメータは、プラグインが OpenSearch Serverless で動作するために必要です。

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

    次の設定ファイルの例では、S3 バケット内のファイルから入力を受け取り、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. 次に、次のように新しい設定で Logstash を実行してプラグインをテストします。

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

Python

以下のサンプルコードでは、Python 用の opensearch-py クライアントを使用して、指定された OpenSearch Serverless コレクションへのセキュアな接続の確立、単一のインデックスの作成、そのインデックスの検索を実行します。region および host の値を指定する必要があります。

OpenSearch Service ドメインとの重要な違いは、サービス名です (es ではなく aoss)。

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

opensearch-aws-sigv4 gem は、追加の設定なしで OpenSearch Serverless とOpenSearch Service へのアクセスを提供します。これには、opensearch-ruby クライアントのすべての機能が備わっています。クライアントがこの gem の依存関係であるためです。

Sigv4 署名者をインスタンス化するときは、aoss をサービス名として指定します。

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)

他のクライアントを使用した HTTP リクエストの署名

以下の要件は、別のクライアントを使用して HTTP リクエストを構築する場合に、OpenSearch Serverless コレクションへのリクエストに署名するときに適用されます。

  • サービス名を aoss として指定する必要があります。

  • x-amz-content-sha256 ヘッダーは、すべての AWS 署名バージョン 4 リクエストに必要です。これにより、リクエストペイロードのハッシュが指定されます。リクエストペイロードがある場合は、その値をセキュアハッシュアルゴリズム (SHA) 暗号化ハッシュ (SHA256) に設定します。リクエストペイロードがない場合は、値を e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855 に設定します。これは空の文字列のハッシュです。

cURL でのインデックス作成

次のリクエスト例では、クライアント URL リクエストライブラリ (cURL) を使用して、コレクション内の movies-index という名前のインデックスに 1 つのドキュメントを送信します。

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"}'

Postman でのインデックス作成

次の図は、Postman を使用してコレクションにリクエストを送信する方法を示しています。認証の手順については、「Postman で AWS の署名による認証ワークフローによる認証」を参照してください。

JSON response showing creation of a "movies-index" with successful result and no shards.