Amazon OpenSearch 인제스티션의 파이프라인 기능 개요

Amazon OpenSearch Ingestion은 소스, 버퍼, 0개 이상의 프로세서, 하나 이상의 싱크로 구성된 파이프라인을 프로비저닝합니다. 수집 파이프라인은 데이터 엔진인 Data Prepper에 의해 구동됩니다. 파이프라인의 다양한 구성 요소에 대한 개요는 주요 개념(을)를 참조하세요.

다음 섹션에서는 Amazon OpenSearch Ingestion에서 가장 일반적으로 사용되는 몇 가지 기능에 대한 개요를 제공합니다.

참고

이 목록은 파이프라인에서 사용할 수 있는 기능 중 전체 목록이 아닙니다. 사용 가능한 모든 파이프라인 기능에 대한 포괄적인 설명서는 Data Prepper 설명서를 참조하세요. 참고로 OpenSearch 인제션은 사용할 수 있는 플러그인과 옵션에 몇 가지 제약을 가합니다. 자세한 정보는 Amazon OpenSearch 통합 파이프라인에 지원되는 플러그인 및 옵션을 참조하세요.

영구 버퍼링

영구 버퍼는 데이터에 내구성을 더하기 위해 여러 가용 영역의 디스크 기반 버퍼에 데이터를 저장합니다. 독립형 버퍼를 설정할 필요 없이 영구 버퍼링을 사용하여 지원되는 모든 푸시 기반 소스의 데이터를 수집할 수 있습니다. 여기에는 HTTP와 로그, 트레이스, 메트릭의 OpenTelemetry 소스가 포함됩니다.

영구 버퍼링을 활성화하려면 파이프라인을 생성하거나 업데이트할 때 영구 버퍼 활성화를 선택합니다. 자세한 내용은 을 참조하십시오Amazon 통합 OpenSearch 파이프라인 생성. OpenSearch 인제션은 파이프라인에 지정한 통합 OpenSearch 컴퓨팅 유닛 (통합 OCU) 에 따라 필요한 버퍼링 용량을 자동으로 결정합니다.

기본적으로 파이프라인은 를 사용하여 버퍼 데이터를 암호화합니다. AWS 소유 키 이러한 파이프라인에는 파이프라인 역할에 대한 추가 권한이 필요하지 않습니다. 또는 고객 관리 키를 지정하고 파이프라인 역할에 다음 IAM 권한을 추가할 수 있습니다.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "KeyAccess",
            "Effect": "Allow",
            "Action": [
              "kms:Decrypt",
              "kms:GenerateDataKeyWithoutPlaintext"
            ],
            "Resource": "arn:aws:kms:{region}:{aws-account-id}:key/1234abcd-12ab-34cd-56ef-1234567890ab"
        }
    ]
}

자세한 내용은 AWS Key Management Service 개발자 안내서의 고객 관리형 키를 참조하십시오.

참고

영구 버퍼링을 비활성화하는 경우 파이프라인이 완전히 인 메모리 버퍼링에서 실행되도록 업데이트됩니다.

최대 요청 페이로드 크기 조정

파이프라인에 영구 버퍼링을 활성화하는 경우 최대 요청 페이로드 크기는 기본적으로 1MB입니다. 기본값이 최상의 성능을 제공합니다. 그러나 클라이언트가 1MB를 초과하는 요청을 보내는 경우 이 값을 늘릴 수 있습니다. 최대 페이로드 크기를 조정하려면 소스 구성 내에서 max_request_length 옵션을 설정합니다. 영구 버퍼링과 마찬가지로 이 옵션은 HTTP 및 로그, 추적 및 OpenTelemetry 지표의 소스에만 지원됩니다.

max_request_length옵션의 유효한 값은 1메가바이트, 1.5메가바이트, 2메가바이트, 2.5메가바이트, 3메가바이트, 3.5메가바이트, 4메가바이트 뿐입니다. 다른 값을 지정하면 오류가 발생합니다.

다음 예제는 파이프라인 구성 내에서 최대 페이로드 크기를 구성하는 방법을 보여줍니다.

...
log-pipeline:
  source:
    http:
      path: "/${pipelineName}/logs"
      max_request_length: 4mb
  processor:
...

파이프라인에 영구 버퍼링을 활성화하지 않는 경우 max_request_length 옵션 값은 모든 소스에 대해 기본적으로 10MB로 설정되며 수정할 수 없습니다.

분할

OpenSearch 수신 이벤트를 하위 파이프라인으로 분할하도록 통합 파이프라인을 구성하여 동일한 수신 이벤트에 대해 다양한 유형의 처리를 수행할 수 있습니다.

다음 예제 파이프라인은 수신 이벤트를 두 개의 하위 파이프라인으로 분할합니다. 각 하위 파이프라인은 자체 프로세서를 사용하여 데이터를 보강하고 조작한 다음 데이터를 다른 인덱스로 보냅니다. OpenSearch

version: "2"
log-pipeline:
  source:
    http:
    ...
  sink:
    - pipeline:
        name: "logs_enriched_one_pipeline"
    - pipeline:
        name: "logs_enriched_two_pipeline"

logs_enriched_one_pipeline:
  source:
    log-pipeline
  processor:
   ...
  sink:
    - opensearch:
        # Provide a domain or collection endpoint
        # Enable the 'serverless' flag if the sink is an OpenSearch Serverless collection
        aws:
          ...
        index: "enriched_one_logs"

logs_enriched_two_pipeline:
  source:
    log-pipeline
  processor:
   ...
  sink:
    - opensearch:
        # Provide a domain or collection endpoint
        # Enable the 'serverless' flag if the sink is an OpenSearch Serverless collection
        aws:
          ...
          index: "enriched_two_logs"

Chaining

데이터 처리 및 보강을 청크 단위로 수행하기 위해 여러 하위 파이프라인을 함께 연결할 수 있습니다. 즉, 수신 이벤트를 하나의 하위 파이프라인에서 특정 처리 기능으로 보강한 다음 다른 하위 파이프라인으로 전송하여 다른 프로세서로 추가 보강하고 마지막으로 해당 싱크로 보낼 수 있습니다. OpenSearch

다음 예제에서 log_pipeline 하위 파이프라인은 들어오는 로그 이벤트를 프로세서 집합으로 보강한 다음 이라는 인덱스로 이벤트를 보냅니다. OpenSearch enriched_logs 파이프라인은 동일한 이벤트를 하위 파이프라인으로 전송합니다. log_advanced_pipeline 하위 파이프라인은 해당 이벤트를 처리하여 이름이 지정된 다른 인덱스로 보냅니다. OpenSearch enriched_advanced_logs

version: "2"
log-pipeline:
  source:
    http:
    ...
  processor:
    ...
  sink:
    - opensearch:
        # Provide a domain or collection endpoint
        # Enable the 'serverless' flag if the sink is an OpenSearch Serverless collection
        aws:
          ...
          index: "enriched_logs"
    - pipeline:
        name: "log_advanced_pipeline"

log_advanced_pipeline:
  source:
    log-pipeline
  processor:
   ...
  sink:
    - opensearch:
        # Provide a domain or collection endpoint
        # Enable the 'serverless' flag if the sink is an OpenSearch Serverless collection
        aws:
          ...
          index: "enriched_advanced_logs"

배달 못한 편지 대기열

DLQ(Dead Letter Queue)는 파이프라인이 싱크에 기록하지 못하는 이벤트의 대상입니다. 통합 OpenSearch 시 DLQ로 사용할 적절한 쓰기 권한이 있는 Amazon S3 버킷을 지정해야 합니다. 파이프라인 내의 모든 싱크에 DLQ 구성을 추가할 수 있습니다. 파이프라인에서 쓰기 오류가 발생하면 구성된 S3 버킷에 DLQ 객체가 생성됩니다. DLQ 객체는 JSON 파일 내에 실패한 이벤트의 배열로 존재합니다.

다음 조건 중 하나가 충족되면 파이프라인이 DLQ에 이벤트를 기록합니다.

• max_retries OpenSearch 싱크용 용량이 모두 소진되었습니다. OpenSearch 이 옵션을 사용하려면 최소 16개가 필요합니다.

• 오류 상태로 인해 싱크에서 이벤트가 거부됩니다.

구성

하위 파이프라인의 DLQ(Dead Letter Queue)를 구성하려면 opensearch 싱크 구성 내에서 dlq 옵션을 지정하세요.

apache-log-pipeline:
  ...
  sink:
    opensearch:
      dlq:
        s3:
          bucket: "my-dlq-bucket"
          key_path_prefix: "dlq-files"
          region: "us-west-2"
          sts_role_arn: "arn:aws:iam::123456789012:role/dlq-role"

이 S3 DLQ에 기록되는 파일은 다음과 같은 이름 지정 패턴을 갖습니다.

dlq-v${version}-${pipelineName}-${pluginId}-${timestampIso8601}-${uniqueId}

자세한 내용은 DLQ(Dead Letter Queue)를 참조하세요.

sts_role_arn 구성 지침은 DLQ(Dead Letter Queue)에 쓰기 섹션을 참조하세요.

예

다음 예제 DLQ 파일을 고려하세요.

dlq-v2-apache-log-pipeline-opensearch-2023-04-05T15:26:19.152938Z-e7eb675a-f558-4048-8566-dac15a4f8343

다음은 싱크에 기록되지 않고 추가 분석을 위해 DLQ S3 버킷으로 전송되는 데이터의 예입니다.

Record_0	
pluginId            "opensearch"
pluginName          "opensearch"
pipelineName        "apache-log-pipeline"
failedData	
index		  "logs"
indexId		 null
status		  0
message		"Number of retries reached the limit of max retries (configured value 15)"
document	
log		    "sample log"
timestamp	    "2023-04-14T10:36:01.070Z"

Record_1	
pluginId            "opensearch"
pluginName          "opensearch"
pipelineName        "apache-log-pipeline"
failedData	
index               "logs"
indexId		 null
status		  0
message		"Number of retries reached the limit of max retries (configured value 15)"
document	
log                 "another sample log"
timestamp           "2023-04-14T10:36:01.071Z"

인덱스 관리

Amazon OpenSearch Ingestion에는 다음을 비롯한 다양한 인덱스 관리 기능이 있습니다.

인덱스 생성

파이프라인 싱크에 인덱스 이름을 지정할 수 있으며, OpenSearch Ingestion은 파이프라인을 프로비저닝할 때 인덱스를 생성합니다. 인덱스가 이미 있는 경우 파이프라인은 해당 인덱스를 사용하여 수신하는 이벤트를 인덱싱합니다. 파이프라인을 중지했다가 다시 시작하거나 YAML 구성을 업데이트하면 파이프라인은 새 인덱스가 아직 없는 경우 새 인덱스를 만들려고 시도합니다. 파이프라인은 인덱스를 삭제할 수 없습니다.

다음 예제 싱크는 파이프라인이 프로비저닝될 때 두 개의 인덱스를 생성합니다.

sink:
  - opensearch:
      index: apache_logs
  - opensearch:
      index: nginx_logs

인덱스 이름 및 패턴 생성

수신 이벤트 필드의 변수를 사용하여 동적 인덱스 이름을 생성할 수 있습니다. 싱크 구성에서는 형식 string${}을 사용하여 문자열 보간을 알리고 JSON 포인터를 사용하여 이벤트에서 필드를 추출합니다. index_type의 옵션은 custom 또는 management_disabled입니다. OpenSearch 도메인과 OpenSearch 서버리스 컬렉션의 custom 경우 index_type 기본값이 management_disabled 사용되므로 설정하지 않은 상태로 둘 수 있습니다.

예를 들어 다음 파이프라인은 수신 이벤트에서 metadataType 필드를 선택하여 인덱스 이름을 생성합니다.

pipeline:
  ...
  sink:
    opensearch:
      index: "metadata-${metadataType}"

다음 구성은 매일 또는 1시간마다 새 인덱스를 계속 생성합니다.

pipeline:
  ...
  sink:
    opensearch:
      index: "metadata-${metadataType}-%{yyyy.MM.dd}"

pipeline:
  ...
  sink:
    opensearch:
      index: "metadata-${metadataType}-%{yyyy.MM.dd.HH}"

인덱스 이름은 my-index-%{yyyy.MM.dd}와 같은 접미사로 날짜-시간 패턴을 포함하는 일반 문자열일 수도 있습니다. 싱크는 데이터를 로 OpenSearch 전송할 때 날짜-시간 패턴을 UTC 시간으로 대체하고 각 날짜에 대해 새 인덱스 (예:) 를 생성합니다. my-index-2022.01.25 자세한 내용은 클래스를 참조하십시오. DateTimeFormatter

이 인덱스 이름은 my-${index}-name(와)과 같은 접미사로 날짜-시간 패턴을 포함 또는 미포함하는 문자열 형식일 수도 있습니다. 싱크는 로 OpenSearch 데이터를 전송할 때 해당 "${index}" 부분을 처리 중인 이벤트의 값으로 대체합니다. 형식이 "${index1/index2/index3}"인 경우 필드 index1/index2/index3가 이벤트의 값으로 대체합니다.

문서 ID 생성

파이프라인은 문서를 인덱싱하는 동안 문서 ID를 생성할 수 있습니다. OpenSearch 수신 이벤트 내의 필드에서 이러한 문서 ID를 유추할 수 있습니다.

이 예제에서는 수신 이벤트의 uuid 필드를 사용하여 문서 ID를 생성합니다.

pipeline:
  ...
  sink:
    opensearch:
      index_type: custom
      index: "metadata-${metadataType}-%{yyyy.MM.dd}" 
      document_id_field: "uuid"   

다음 예제에서 항목 추가 프로세서는 수신 이벤트의 uuid 및 other_field 필드를 병합하여 문서 ID를 생성합니다.

create 작업을 수행하면 ID가 동일한 문서를 덮어쓰지 않습니다. 파이프라인은 재시도 또는 DLQ 이벤트 없이 중복 문서를 삭제합니다. 기존 문서를 업데이트하지 않는 것이 목적이므로 이 작업을 사용하는 파이프라인 작성자에게는 이 작업을 예상하는 것이 합리적입니다.

pipeline:
  ...
  processor:
   - add_entries:
      entries:
        - key: "my_doc_id_field"
          format: "${uuid}-${other_field}"
  sink:
    - opensearch:
       ...
       action: "create"
       document_id_field: "my_doc_id_field"

이벤트의 문서 ID를 하위 객체의 필드로 설정하고 싶을 수도 있습니다. 다음 예제에서 OpenSearch 싱크 플러그인은 하위 오브젝트를 info/id 사용하여 문서 ID를 생성합니다.

sink:
  - opensearch:
       ...
       document_id_field: info/id

다음 이벤트가 발생하면 파이프라인은 json001로 설정된 _id 필드로 문서를 생성합니다.

{
   "fieldA":"arbitrary value",
   "info":{
      "id":"json001",
      "fieldA":"xyz",
      "fieldB":"def"
   }
}

라우팅 ID 생성

OpenSearch 싱크 플러그인의 routing_field 옵션을 사용하여 문서 라우팅 속성 (_routing) 의 값을 수신 이벤트의 값으로 설정할 수 있습니다.

라우팅은 JSON 포인터 구문을 지원하므로 최상위 필드뿐 아니라 중첩된 필드도 사용할 수 있습니다.

sink:
  - opensearch:
       ...
       routing_field: metadata/id
       document_id_field: id

다음 이벤트가 발생하면 플러그인은 abcd로 설정된 _routing 필드로 문서를 생성합니다.

{
   "id":"123",
   "metadata":{
      "id":"abcd",
      "fieldA":"valueA"
   },
   "fieldB":"valueB"
}

파이프라인이 인덱스 생성 중에 사용할 수 있는 인덱스 템플릿을 만드는 방법에 대한 지침은 인덱스 템플릿을 참조하세요.

E nd-to-end 승인

OpenSearch Ingestion은 승인을 사용하여 무상태 파이프라인의 소스에서 싱크까지 전달되는 데이터를 추적하여 데이터의 내구성과 신뢰성을 보장합니다. end-to-end 현재는 S3 소스 플러그인만 승인을 지원합니다. end-to-end

end-to-end 승인 기능을 사용하면 파이프라인 소스 플러그인이 승인 세트를 생성하여 이벤트 배치를 모니터링합니다. 이벤트가 싱크로 성공적으로 전송되면 긍정적인 승인을 받고, 싱크로 전송할 수 없는 이벤트가 있을 때는 부정적인 승인을 받습니다.

파이프라인 구성 요소에 장애 또는 충돌이 발생하거나 소스가 승인을 받지 못하는 경우 소스는 제한 시간이 초과되어 실패 재시도 또는 로깅과 같은 필요한 조치를 취합니다. 파이프라인에 싱크가 여러 개 있거나 하위 파이프라인이 여러 개 구성된 경우 이벤트가 모든 하위 파이프라인의 모든 싱크에 전송된 후에만 이벤트 수준 승인이 전송됩니다. 싱크에 DLQ가 구성된 경우 end-to-end 승인 메시지는 DLQ에 기록된 이벤트도 추적합니다.

end-to-end 승인을 활성화하려면 소스 구성 내에 옵션을 포함하세요. acknowledgments

s3-pipeline:
  source:
    s3:
      acknowledgments: true
...

소스 백 프레셔

파이프라인이 데이터를 처리하느라 바쁠 때, 싱크가 일시적으로 다운되거나 데이터 수집 속도가 느릴 경우 역압이 발생할 수 있습니다. OpenSearch 인제스트는 파이프라인이 사용하는 소스 플러그인에 따라 백 프레셔를 처리하는 방법이 다릅니다.

HTTP 소스

HTTP 소스 플러그인을 사용하는 파이프라인은 혼잡한 파이프라인 구성 요소에 따라 배압을 다르게 처리합니다.

• 버퍼 - 버퍼가 가득 차면 파이프라인이 오류 코드 408과 함께 HTTP 상태 REQUEST_TIMEOUT를 소스 엔드포인트로 반환하기 시작합니다. 버퍼가 비워지면 파이프라인은 HTTP 이벤트 처리를 다시 시작합니다.

• 소스 스레드 — 모든 HTTP 소스 스레드가 요청을 실행하는 중이고 처리되지 않은 요청 대기열 크기가 허용된 최대 요청 수를 초과하면 파이프라인은 오류 코드 429와 함께 HTTP 상태 TOO_MANY_REQUESTS를 소스 엔드포인트로 반환하기 시작합니다. 요청 대기열이 최대 허용 대기열 크기 아래로 떨어지면 파이프라인은 요청 처리를 다시 시작합니다.

OTel 소스

OpenTelemetry 소스 (OTel 로그, oTel 메트릭, OTel 추적) 를 사용하는 파이프라인의 버퍼가 가득 차면 파이프라인은 오류 코드 408과 REQUEST_TIMEOUT 함께 HTTP 상태를 소스 엔드포인트에 반환하기 시작합니다. 버퍼가 비워지면 파이프라인은 이벤트 처리를 다시 시작합니다.

S3 소스

S3 소스가 있는 파이프라인의 버퍼가 가득 차면 파이프라인의 SQS 알림 처리가 중지됩니다. 버퍼가 비워지면 파이프라인에서 알림 처리를 다시 시작합니다.

싱크가 다운되거나 데이터를 수집할 수 없고 소스에 대한 확인이 활성화된 경우 파이프라인은 모든 싱크로부터 성공적인 end-to-end 승인을 받을 때까지 SQS 알림 처리를 중단합니다.