$emit 애그리게이션 단계(스트림 처리)

문자열

필수 사항

데이터를 수집할 연결의 이름은 연결 레지스트리에 나타나는 대로입니다.

문자열 | 표현식

필수 사항

메시지를 보낼 Apache Kafka 주제 의 이름입니다.

문서

옵션

다양한 기본값을 재정의하는 필드가 포함된 문서입니다.

int

옵션

성공적인 $emit 작업을 위해 Apache Kafka 클러스터 에서 필요한 승인의 수입니다.

기본값은 all입니다. Atlas Stream Processing은 다음 값을 지원합니다.

• -1

• 0

• 1

• all

문자열

옵션

생성자가 생성한 모든 데이터에 대한 압축 유형입니다. 기본값은 없음(즉, 압축하지 않음)입니다. 유효한 값은 다음과 같습니다.

• none

• gzip

• snappy

• lz4

• zstd

압축은 전체 데이터 배치에 사용되므로 배치의 효율성이 압축 비율에 영향을 미칩니다. 배치 처리가 많을수록 더 나은 압축이 가능합니다.

문자열

옵션

날짜 값의 날짜 형식입니다. 유효한 값은 다음과 같습니다.

• default - outputFormat의 기본값을 사용합니다.

• ISO8601 - 날짜를 ISO8601 형식의 문자열로 변환하며, 여기에는 밀리초 정밀도(YYYY-MM-DDTHH:mm:ss.sssZ)가 포함됩니다.

예를 들면 다음과 같습니다.

다음 입력을 고려하세요.

{ "flightTime" :  ISODate('2025-01-10T20:17:38.387Z') }

$emit.config.dateFormat 이 default로 설정된 경우 출력은 다음과 유사하게 표시됩니다.

{ "flightTime" :  {$date :"2025-01-10T20:17:38.387Z"}}

$emit.config.dateFormat 이 ISO8601로 설정된 경우 출력은 다음과 유사하게 표시됩니다.

{ "flightTime" :  "2025-01-10T20:17:38.387Z" }

표현식

옵션

출력 메시지에 추가할 헤더입니다. 표현식은 객체 또는 배열로 평가되어야 합니다.

표현식이 객체로 평가되는 경우, Atlas Stream Processing은 해당 객체의 각 키-값 쌍에서 헤더를 구성합니다. 여기서 키는 헤더 이름이고 값은 헤더 값입니다.

표현식이 배열로 평가되는 경우 키-값 쌍 객체배열 형식을 취해야 합니다. 예시:

[
  {k: "name1", v: ...},
  {k: "name2", v: ...},
  {k: "name3", v: ...}
]

Atlas Stream Processing은 배열의 각 객체로부터 헤더를 구성하며 이때 키는 헤더 이름이고 값은 헤더 값입니다. Atlas Stream Processing은 다음 유형의 헤더 값을 지원합니다.

• binData

• string

• object

• int

• long

• double

• null

객체 | 문자열 | 표현식

옵션

Apache Kafka 메시지 키로 평가되는 표현식입니다.

config.key를 지정하는 경우 config.keyFormat을 지정해야 합니다.

문자열

조건부

Apache Kafka 키 데이터를 직렬화하는 데 사용되는 데이터 유형입니다. 다음 값 중 하나여야 합니다.

• "binData"

• "string"

• "json"

• "int"

• "long"

기본값은 binData입니다. config.key를 지정하는 경우 config.keyFormat을 지정해야 합니다. 문서의 config.key가 지정된 데이터 타입으로 성공적으로 직렬화되지 않으면 Atlas Stream Processing은 이를 데드 레터 큐로 보냅니다.

문자열

옵션

Apache Kafka 로 메시지를 보낼 때 사용할 JSON 형식입니다.. 다음 값 중 하나여야 합니다.

• "relaxedJson"

• "canonicalJson"

기본값은 "relaxedJson"입니다.

표현식

옵션

null 을 Kafka 로 내보낼 시기를 결정하는 표현식입니다. 표현식 부울 true 또는 false로 평가되어야 합니다. 특정 문서 에 대해 표현식 true 로 평가되면 Atlas Stream Processing 대신 Kafka 싱크로 null 를 방출합니다. 표현식 $emit false로 평가되면 Atlas Stream Processing 문서 내보냅니다.

표현식 부울 값으로 평가되지 않거나 평가할 수 없는 경우 Atlas Stream Processing 문서 DLQ에 씁니다.

$emit.config.key 및 $emit.config.keyFormat 값을 제공하는 경우 이 설정을 사용하여 주제 압축 활성화 수 있습니다. 이러한 값을 제공하지 않으면 이 표현식 true로 평가될 때 Atlas Stream Processing 여전히 null 를 방출하지만, 이로 인해 Kafka 주제 압축 트리거하다 되지는 않습니다.

문자열

필수 사항

데이터를 쓰기 위한 연결의 이름은 연결 레지스트리에 나타나는 대로입니다.

문자열

필수 사항

데이터를 쓰기할 S3 버킷의 이름입니다.

문자열

옵션

대상 버킷이 위치한 AWS 리전의 이름입니다. 스트림 처리 인스턴스를 AWS 리전에 호스팅하면, 이 매개변수는 기본적으로 해당 리전으로 설정됩니다. 그 외의 경우, 스트림 처리 인스턴스 호스트 리전과 가장 가까운 AWS 리전이 기본값으로 설정됩니다.

문자열 | 표현식

필수 사항

S3 버킷에 기록된 객체 키의 접두사입니다. 리터럴 접두사 문자열이거나 문자열로 평가되는 표현식 이어야 합니다.

문서

옵션

기본값을 재정의하는 추가 매개변수가 포함된 문서입니다.

문서

옵션

쓰기 동작을 제어하는 추가 매개변수가 포함된 문서입니다. 이러한 매개변수는 어떤 임계값이 먼저 충족되는지에 따라 쓰기 동작을 트리거합니다.

예를 들어, 수집된 문서가 config.writeOptions.count 임계값에 도달했지만 config.writeOptions.interval 임계값에는 도달하지 않은 경우 스트림 프로세서는 config.writeOptions.count 임계값에 따라 이러한 문서를 S3로 계속 내보냅니다.

integer

옵션

S3에 기록되는 각 파일에 그룹화할 문서 수입니다.

integer

옵션

파일이 S3에 기록되기 전까지 누적되어야 하는 최소 바이트 수를 지정합니다. 바이트 수는 파이프라인에서 수집된 BSON 문서의 크기에 따라 결정되며, 최종 출력 파일의 크기는 고려되지 않습니다.

문서

옵션

size 및 units의 조합으로 문서 대량 쓰기를 위한 타이머를 지정합니다.

기본값은 1분입니다. unit에 대해 size를 0으로 설정할 수 없습니다. 최대 간격은 7일입니다.

integer

조건부

writeOptions.interval.units로 지정된 단위 수가 경과하면, 스트림 프로세서가 S3에 문서를 대량으로 기록합니다.

기본값은 1입니다. size를 0으로 설정할 수 없습니다. writeOptions.interval를 정의하면 이 매개변수도 반드시 정의해야 합니다.

문자열

조건부

대량 쓰기 타이머를 측정할 시간의 단위입니다. 이 매개변수는 다음 값을 지원합니다.

• ms

• second

• minute

• hour

• day

기본값은 minute입니다. writeOptions.interval을 정의하면 이 매개변수도 반드시 정의해야 합니다.

문자열

옵션

내보낸 파일의 각 항목 사이의 구분자입니다.

기본값은 \n입니다.

문자열

옵션

S3에 기록된 JSON의 출력 형식을 지정합니다. 다음 값 중 하나여야 합니다.

• "basicJson"

• "canonicalJson"

• "relaxedJson"

기본값은 "relaxedJson"입니다.

자세한 내용을 보려면 기본 JSON을 참조하세요.

문자열

옵션

날짜 값의 날짜 형식입니다. 유효한 값은 다음과 같습니다.

• default - outputFormat의 기본값을 사용합니다.

• ISO8601 - 날짜를 ISO8601 형식의 문자열로 변환하며, 여기에는 밀리초 정밀도(YYYY-MM-DDTHH:mm:ss.sssZ)가 포함됩니다.

예를 들어, 파이프라인에 다음 기록을 추가하는 경우:

{ "flightTime" :  ISODate('2025-01-10T20:17:38.387Z') }

만약 $emit.config.dateFormat이 default로 설정된 경우 출력은 다음과 유사하게 나타납니다.

{ "flightTime" :  {$date :"2025-01-10T20:17:38.387Z"}}

$emit.config.dateFormat 이 ISO8601로 설정된 경우 출력은 다음과 유사하게 표시됩니다.

{ "flightTime" :  "2025-01-10T20:17:38.387Z" }

문자열

옵션

사용할 압축 알고리즘의 이름입니다. 다음 값 중 하나여야 합니다.

• "gzip"

• "snappy"

문자열

조건부

내보내는 메시지에 적용할 압축 단계입니다. 1-9를 포함하는 값을 지원하며, 값이 클수록 압축률이 더 높아집니다.

기본값은 6입니다.

이 매개변수는 gzip에만 필요하며 해당 방식에만 적용됩니다. config.compression을snappy로 설정하면 이 매개변수를 설정해도 효과가 없습니다.

{ "binary": { "$binary": { "base64": "gf1UcxdHTJ2HQ/EGQrO7mQ==", "subType": "00" }}}

{ "binary": "gf1UcxdHTJ2HQ/EGQrO7mQ=="}

{ "date": { "$date": "2024-10-24T18:07:29.636Z"}}

{ "date": 1729625275856}

{ "decimal": { "$numberDecimal": "9.9" }}

{ "decimal": "9.9" }

{ "timestamp": { "$timestamp": { "t": 1729793249, "i": 1 }}}

{ "timestamp": 1729793249000}

{ "_id": { "$oid": "671a8ce1497407eff0e17cba" }}

{ "_id": "6717fcbba18c8a8f74b6d977" }

{ "negInf": { "$numberDouble": "-Infinity" }}

{ "negInf": "-Infinity" }

{ "posInf": { "$numberDouble": "Infinity" }}

{ "posInf": "Infinity" }

{ "regex": { "$regularExpression": { "pattern": "ab+c", "options": "i" }}}

{ "regex": { "pattern": "ab+c", "options": "i" }}

{ "uuid": { "$binary": { "base64": "Kat+fHk6RkuAmotUmsU7gA==", "subType": "04" }}}

{ "uuid": "420b7ade-811a-4698-aa64-c8347c719cf1"}