본문 바로가기
WIKI 기술 지식 베이스

S3 sink

원문 보기 위키 갱신

출처: 문서

본문

S3 sink

s3 싱크는 OpenSearch Data Prepper 이벤트 배치를 Amazon Simple Storage Service(Amazon S3) 객체에 저장하고 기록해요. 설정된 codec이 s3 싱크가 데이터를 Amazon S3에 직렬화하는 방식을 결정해요.

s3 싱크는 이벤트를 배치로 묶을 때 다음 형식을 사용해요:

${pathPrefix}events-%{yyyy-MM-dd'T'HH-mm-ss'Z'}-${currentTimeInNanos}-${uniquenessId}.${codecSuppliedExtension}

{% include copy.html %}

객체 배치가 Amazon S3에 기록되면 객체는 다음과 비슷한 형식으로 작성돼요:

my-logs/2023/06/09/06/events-2023-06-09T06-00-01-1686290401871214927-ae15b8fa-512a-59c2-b917-295a0eff97c8.json

{% include copy.html %}

객체 구성 방법에 대한 자세한 내용은 Object key를 참조하세요.

사용 방법

다음 예제는 s3 싱크로 구성된 파이프라인을 만들어요. 파이프라인의 이벤트 및 크기 임계값을 커스터마이징하는 추가 옵션을 포함하고, 코덱 타입을 ndjson으로 설정해요:

pipeline:
  ...
  sink:
    - s3:
        aws:
          region: us-east-1
          sts_role_arn: arn:aws:iam::123456789012:role/Data-Prepper
        max_retries: 5
        bucket: bucket_name
        object_key:
          path_prefix: my-logs/%{yyyy}/%{MM}/%{dd}/
        threshold:
          event_count: 10000
          maximum_size: 50mb
          event_collect_timeout: 15s
        codec:
          ndjson:
        buffer_type: in_memory

{% include copy.html %}

IAM 권한

s3 싱크를 사용하려면 AWS Identity and Access Management(IAM)를 구성해서 Data Prepper에 Amazon S3 쓰기 권한을 부여해야 해요. 다음 JSON 구성과 비슷한 구성을 사용할 수 있어요:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "s3-access",
            "Effect": "Allow",
            "Action": [
              "s3:PutObject"
            ],
            "Resource": "arn:aws:s3:::<YOUR-BUCKET>/*"
        }
    ]
}

{% include copy.html %}

교차 계정 S3 액세스

Data Prepper가 S3 버킷에서 데이터를 가져올 때는 버킷 소유자 조건을 사용해 버킷 소유권을 확인해요.

기본적으로 S3 싱크는 bucket_owners를 필요로 하지 않아요. bucket_owners가 설정되어 있고 어떤 버킷이 매핑된 구성 중 어디에도 포함되지 않으면, default_bucket_owner는 aws.sts_role_arn의 계정 ID를 기본값으로 사용해요. bucket_owners와 default_bucket_owner를 둘 다 설정하고 이 설정들을 함께 적용할 수도 있어요.

서로 다른 계정 연관 관계를 가진 여러 S3 버킷에서 데이터를 수집할 때는 다음 조건에 따라 교차 계정 S3 액세스를 위해 Data Prepper를 구성하세요:

  • 같은 계정에 속한 S3 버킷이라면 default_bucket_owner를 해당 계정의 ID로 설정하세요.
  • 여러 계정에 속한 S3 버킷이라면 bucket_owners 맵을 사용하세요.

bucket_owners 맵은 여러 계정에 속한 버킷들의 계정 ID를 지정해요. 예를 들어 다음 구성에서는 my-bucket-01의 소유자가 123456789012이고 my-bucket-02의 소유자는 999999999999예요:

sink:
  - s3:
      default_bucket_owner: 111111111111
      bucket_owners:
        my-bucket-01: 123456789012
        my-bucket-02: 999999999999

{% include copy.html %}

구성

s3 싱크를 커스터마이징할 때 다음 옵션을 사용하세요.

옵션 필수 여부 타입 설명
bucket 예 String 싱크의 S3 버킷 이름을 지정해요. Data Prepper 표현식을 사용한 동적 버킷 네이밍을 지원하며, 예를 들어 test-${/bucket_id} 같은 형태예요. 동적 버킷에 접근할 수 없고 default_bucket이 설정되어 있지 않으면 객체 데이터는 버려져요.
default_bucket 아니요 String bucket의 동적 버킷에 접근할 수 없을 때 사용하는 정적 버킷이에요.
bucket_owners 아니요 Map 교차 계정 액세스를 위한 버킷 이름과 계정 소유자 ID의 맵이에요. 자세한 내용은 교차 계정 S3 액세스를 참조하세요.
default_bucket_owner 아니요 String S3 버킷 소유자의 AWS 계정 ID예요. 자세한 내용은 교차 계정 S3 액세스를 참조하세요.
codec 예 Codec S3 객체의 데이터를 직렬화해요.
aws 예 AWS AWS 구성이에요. 자세한 내용은 aws를 참조하세요.
threshold 예 Threshold 객체를 S3에 쓰기 위한 조건이에요.
aggregate_threshold 아니요 Aggregate threshold 동적 path_prefix로 객체를 플러시하기 위한 조건이에요.
object_key 아니요 Object key 객체 저장을 위한 path_prefix와 file_pattern을 설정해요. 파일 패턴은 events-%{yyyy-MM-dd'T'hh-mm-ss}예요. 기본적으로 이 객체들은 버킷의 루트 디렉터리에서 찾을 수 있어요. path_prefix는 설정 가능해요.
compression 아니요 String 압축 알고리즘이에요. none, gzip, snappy 중 하나예요. 기본값은 none이에요.
buffer_type 아니요 Buffer type 버퍼 타입 구성이에요.
max_retries 아니요 Integer S3 수집 요청의 최대 재시도 횟수예요. 기본값은 5예요.

aws

옵션 필수 여부 타입 설명
region 아니요 String 자격 증명에 사용할 AWS 리전이에요. 기본적으로 리전을 결정하는 표준 SDK 동작을 따라요.
sts_role_arn 아니요 String Amazon Simple Queue Service(Amazon SQS)와 Amazon S3에 대한 요청에서 수임할 AWS Security Token Service(AWS STS) 역할이에요. 기본값은 null이며, 이 경우 자격 증명을 위한 표준 SDK 동작을 사용해요.
sts_header_overrides 아니요 Map 싱크 플러그인에서 IAM 역할이 수임될 때 적용되는 헤더 재정의 맵이에요.
sts_external_id 아니요 String Data Prepper가 역할을 수임할 때 사용하는 AWS STS 외부 ID예요. 자세한 내용은 AWS STS API 레퍼런스의 AssumeRole 아래 ExternalId 섹션을 참조하세요.

Threshold 구성

다음 옵션을 사용해 s3 싱크의 수집 임계값을 설정하세요. 이 조건 중 하나라도 발생하면 Data Prepper가 이벤트를 S3 객체에 기록해요.

옵션 필수 여부 타입 설명
event_count 예 Integer 객체를 S3에 쓰기 전에 누적할 Data Prepper 이벤트 수예요.
maximum_size 아니요 String 객체를 S3에 쓰기 전에 누적할 최대 바이트 수예요. 기본값은 50mb예요.
event_collect_timeout 예 String Data Prepper가 이벤트를 S3에 쓰기 전까지의 최대 시간이에요. 값은 PT2M30S 같은 ISO-8601 기간이나 60s, 1500ms 같은 간단한 표기법 중 하나여야 해요.

Aggregate threshold 구성

다음 옵션을 사용해, 집계된 값이 정의된 임계값을 넘을 때 특정 동작이나 행동을 트리거하는 규칙 또는 제한을 설정하세요.

옵션 필수 여부 타입 설명
flush_capacity_ratio 아니요 Float aggregate_threshold maximum_size에 도달했을 때 강제로 플러시할 그룹의 비율이에요. 비율은 0.0과 1.0 사이의 숫자로 표현해요. 기본값은 0.5예요.
maximum_size 예 String 객체를 강제로 플러시하기 전에 누적할 최대 바이트 수예요. 예를 들어 128mb가 있어요.

Buffer type

buffer_type은 객체를 S3에 쓰기 전에 Data Prepper가 데이터를 임시 저장하는 방식을 결정하는 선택적 구성이에요. 기본값은 in_memory예요.

다음 옵션 중 하나를 사용하세요:

  • in_memory: 레코드를 메모리에 저장해요.
  • local_file: 레코드를 로컬 머신의 파일로 플러시해요. 이 옵션은 머신의 임시 디렉터리를 사용해요.
  • multipart: S3 멀티파트 업로드를 사용해 기록해요. 10 MB마다 하나의 파트로 쓰여요.

Object key 구성

다음 옵션을 사용해 S3에 저장되는 객체의 객체 키가 만들어지는 방식을 정의하세요.

옵션 필수 여부 타입 설명
path_prefix 아니요 String S3에 기록되는 객체에 사용할 S3 키 접두사 경로예요. 날짜-시간 포맷과 Data Prepper 표현식을 사용한 값의 동적 주입을 받아들여요. 예를 들어 /${/my_partition_key}/%{yyyy}/%{MM}/%{dd}/%{HH}/를 사용하면 my_partition_key 값을 기반으로 S3에 시간별 폴더를 만들 수 있어요. 접두사 경로는 /로 끝나야 해요. 기본적으로 Data Prepper는 객체를 S3 버킷 루트에 기록해요.

codec

codec은 s3 소스가 각 S3 객체에 기록되는 데이터를 포맷하는 방식을 결정해요.

avro codec

avro 코덱은 이벤트를 Apache Avro 문서로 기록해요. Avro는 스키마가 필요하기 때문에, 스키마를 직접 정의하거나 Data Prepper가 자동으로 생성하게 할 수 있어요. 스키마를 직접 정의하면 특정 사용 사례에 맞게 조정할 수 있기 때문에 직접 정의하는 것이 권장돼요.

직접 Avro 스키마를 제공하면 그 스키마가 데이터의 최종 구조를 정의해요. Avro 스키마에 매핑되지 않은, 들어오는 이벤트의 추가 값들은 최종 목적지에 포함되지 않아요. Data Prepper는 사용자 정의 Avro 스키마와 include_keys 또는 exclude_keys 싱크 구성 간의 혼동을 피하기 위해, 사용자 정의 스키마와 함께 include_keys나 exclude_keys를 사용하는 것을 허용하지 않아요.

데이터가 균일한 경우라면 스키마를 자동으로 생성할 수도 있어요. 자동 생성된 스키마는 코덱이 수신하는 첫 번째 이벤트를 기반으로 해요. 스키마에는 이 이벤트의 키만 포함되며, 동작하는 스키마를 자동으로 생성하려면 모든 키가 모든 이벤트에 존재해야 해요. 자동 생성된 스키마는 모든 필드를 nullable로 만들어요. 자동 생성된 스키마에 어떤 데이터가 포함될지 제어하려면 include_keys와 exclude_keys 싱크 구성을 사용하세요.

Avro 필드는 누락된 값을 허용할 수 있도록 null 유니언을 사용해야 해요. 그렇지 않으면 모든 필수 필드가 각 이벤트마다 존재해야 해요. non-nullable 필드는 그 필드가 반드시 존재한다고 확신할 때만 사용하세요.

다음 옵션으로 코덱을 구성하세요.

옵션 필수 여부 타입 설명
schema 예 String Avro 스키마 선언이에요. auto_schema가 true로 설정되면 필요하지 않아요.
auto_schema 아니요 Boolean true로 설정하면 첫 번째 이벤트에서 Avro 스키마 선언을 자동으로 생성해요.

ndjson codec

ndjson 코덱은 각 줄을 JSON 객체로 기록해요. ndjson 코덱은 어떤 구성도 받지 않아요.

json codec

json 코덱은 이벤트를 하나의 큰 JSON 파일로 기록해요. 각 이벤트는 JSON 배열 안의 객체로 기록돼요.

다음 옵션으로 코덱을 구성하세요.

옵션 필수 여부 타입 설명
key_name 아니요 String JSON 배열의 키 이름이에요. 기본값은 events예요.

parquet codec

parquet 코덱은 이벤트를 Parquet 파일로 기록해요. 이 코덱을 사용할 때는 buffer_type을 in_memory로 설정하세요.

parquet 코덱은 스키마를 사용해 데이터를 기록해요. Parquet은 Avro 스키마가 필요하기 때문에, 스키마를 직접 정의하거나 Data Prepper가 자동으로 생성하게 할 수 있어요. 스키마를 직접 정의하면 특정 사용 사례에 맞게 조정할 수 있기 때문에 직접 정의하는 것이 권장돼요.

Avro 스키마에 대한 자세한 내용은 Avro codec을 참조하세요.

다음 옵션으로 코덱을 구성하세요.

옵션 필수 여부 타입 설명
schema 예 String Avro 스키마 선언이에요. auto_schema가 true로 설정되면 필요하지 않아요.
auto_schema 아니요 Boolean true로 설정하면 첫 번째 이벤트에서 Avro 스키마 선언을 자동으로 생성해요.

Parquet에서 스키마 설정하기

다음 예제 파이프라인은 VPC Flow Logs용 스키마를 사용해 Parquet 데이터를 Parquet 파일로 기록하도록 s3 싱크를 구성하는 방법을 보여줘요:

pipeline:
  ...
  sink:
    - s3:
        aws:
          region: us-east-1
          sts_role_arn: arn:aws:iam::123456789012:role/Data-Prepper
        bucket: mys3bucket
        object_key:
          path_prefix: vpc-flow-logs/%{yyyy}/%{MM}/%{dd}/%{HH}/
        codec:
          parquet:
            schema: >
              {
                "type" : "record",
                "namespace" : "org.opensearch.dataprepper.examples",
                "name" : "VpcFlowLog",
                "fields" : [
                  { "name" : "version", "type" : ["null", "string"]},
                  { "name" : "srcport", "type": ["null", "int"]},
                  { "name" : "dstport", "type": ["null", "int"]},
                  { "name" : "accountId", "type" : ["null", "string"]},
                  { "name" : "interfaceId", "type" : ["null", "string"]},
                  { "name" : "srcaddr", "type" : ["null", "string"]},
                  { "name" : "dstaddr", "type" : ["null", "string"]},
                  { "name" : "start", "type": ["null", "int"]},
                  { "name" : "end", "type": ["null", "int"]},
                  { "name" : "protocol", "type": ["null", "int"]},
                  { "name" : "packets", "type": ["null", "int"]},
                  { "name" : "bytes", "type": ["null", "int"]},
                  { "name" : "action", "type": ["null", "string"]},
                  { "name" : "logStatus", "type" : ["null", "string"]}
                ]
              }
        threshold:
          event_count: 500000000
          maximum_size: 20mb
          event_collect_timeout: PT15M
        buffer_type: in_memory

{% include copy.html %}

더 알아보기 (Learn more)

함께 읽는 실무 이야기

우리 조직에도 적용하고 싶다면, AI 시스템 구축 서비스를 살펴보세요.