/ /

`$source` 단계(Stream Processing)

정의

$source

The $source stage specifies a connection in the Connection Registry to stream data from. The following connection types are supported:

Apache Kafka 브로커
MongoDB collection change stream
MongoDB database change stream
MongoDB 클러스터 변경 스트림
AWS Kinesis 데이터 스트림
문서 배열
Cron schedule

구문

Apache Kafka 브로커

Apache Kafka 브로커의 스트리밍 데이터로 작업하기 위해 $source 단계의 프로토타입 형식은 다음과 같습니다.

{
  "$source": {
    "connectionName": "<registered-connection>",
    "topic" : ["<source-topic>", ...],
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "partitionIdleTimeout": {
      "size": <duration-number>,
      "unit": "<duration-unit>"
    },
    "schemaRegistry": {
      "connectionName": "<schema-registry-name>",
    },
    "config": {
      "auto_offset_reset": "<start-event>",
      "group_id": "<group-id>",
      "keyFormat": "<deserialization-type>",
      "keyFormatError": "<error-handling>"
    },
  }
}

$source 단계에서는 다음 필드가 있는 문서를 사용합니다.

필드

유형

필요성

설명

connectionName

문자열

필수 사항

데이터를 수집할 연결 레지스트리 에서 연결을 식별하는 레이블입니다.

topic

문자열 또는 문자열 배열

필수 사항

메시지를 스트림할 하나 이상의 Apache Kafka 주제 이름입니다. 둘 이상의 주제 에서 메시지를 스트림 하려면 해당 주제를 배열 로 지정합니다.

timeField

문서

옵션

수신 메시지에 대한 신뢰할 수 있는 타임스탬프를 정의하는 문서입니다.

timeField 을(를) 사용하는 경우 다음 중 하나로 정의해야 합니다.

소스 메시지 필드 인수로 사용하는 a : 표현식 :$toDate 입니다.
소스 메시지 필드 인수로 사용하는 a : 표현식 :$dateFromString 입니다.

timeField 을 선언하지 않으면 Atlas Stream Processing은 소스에서 제공한 메시지 타임스탬프에서 타임스탬프를 생성합니다.

partitionIdleTimeout

문서

옵션

파티션이 워터마크 계산에서 무시되기 전에 유휴 상태로 허용되는 시간을 지정하는 문서입니다.

이 필드는 기본적으로 비활성화되어 있습니다. 유휴 상태로 인해 진행되지 않는 파티션을 처리하려면 이 필드에 값을 설정하세요.

partitionIdleTimeout.size

integer

옵션

파티션 유휴 시간 초과 기간을 지정하는 숫자입니다.

partitionIdleTimeout.unit

문자열

옵션

파티션 유휴 시간 초과 기간의 시간 단위입니다.

unit 의 값은 다음 중 하나일 수 있습니다.

"ms" (밀리초)
"second"
"minute"
"hour"
"day"

schemaRegistry

문서

옵션

Avro 직렬화된 소스에서 읽기를 지원 위해 스키마 레지스트리를 사용할 수 있도록 하는 문서입니다.

이 기능 활성화 하려면 스키마 레지스트리 연결을 만들어야 합니다.

schemaRegistry.connectionName

문자열

조건부

Avro 역직렬화에 사용할 스키마 레지스트리 연결의 이름입니다.

config

문서

옵션

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

config.auto_offset_reset

문자열

옵션

수집을 시작할 Apache Kafka 소스 주제 의 이벤트 지정합니다. auto_offset_reset 는 다음 값을 사용합니다.

end, latest 또는 largest 을 사용하여 애그리게이션이 초기화될 때 주제의 최신 이벤트로부터 수집을 시작합니다.
earliest, beginning 또는 smallest : 주제에서 가장 오래된 이벤트부터 수집을 시작합니다.

기본값은 latest입니다.

config.group_id

문자열

옵션

스트림 프로세서와 연결할 kafka 소비자 그룹의 ID입니다. 생략할 경우, Atlas Stream Processing은 스트림 처리 작업 공간을 다음 형식의 자동 생성 ID와 연결합니다.

asp-${streamProcessorId}-consumer

Atlas Stream Processing 모든 영구 스트림 프로세서에 대해 이 매개변수 값을 자동으로 생성합니다. SP로 정의된 임시 스트림 프로세서의 경우. 프로세스()에서 이 매개변수는 수동으로 정의한 경우에만 설정하다 됩니다.

config.enable_auto_commit

부울

조건부

Kafka 브로커 파티션 오프셋에 대한 커밋 정책을 결정하는 플래그입니다. Atlas Stream Processing 두 가지 커밋 정책을 지원합니다.

이 매개 변수를 true로 설정하다 Atlas Stream Processing $source 단계에서 다음 연산자 로 데이터를 전달할 때마다 오프셋을 커밋합니다.
이 매개 변수를 false로 설정하다 하면 Atlas Stream Processing 체크포인트 취할 때 스트림 프로세서가 파티션 오프셋을 커밋 .

SP로 정의된 임시 스트림 프로세서의 경우. 프로세스()에서 false 을 설정하다 하지 않는 한 이 매개 변수의 group_id 기본값은 입니다. 그렇지 않으면 기본값은 true 입니다.

Kafka $source로 사용할 때 오프셋에 대한 자세한 내용은 Kafka 소스 및 소비자 그룹 오프셋을 참조하세요.

config.keyFormat

문자열

옵션

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

"binData"
"string"
"json"
"int"
"long"

기본값은 binData입니다.

config.keyFormatError

문자열

옵션

Apache Kafka 키 데이터를 역직렬화할 때 발생하는 오류를 처리하다 방법. 다음 값 중 하나여야 합니다.

dlq는, 해당 문서를 데드 레터 큐에 기록합니다.
passThrough- 키 데이터 없이 문서를 다음 단계로 보냅니다.

참고

Atlas Stream Processing에서는 소스 데이터 스트림의 문서가 유효한 json 또는 ejson이어야 합니다. Atlas Stream Processing은 데드 레터 큐를 구성한 경우 이 요구 사항을 충족하지 않는 문서를 데드 레터 큐로 설정합니다.

MongoDB collection change stream

An Atlas collection change stream allows applications to access real-time data changes on a single collection. To learn how to open a change stream against a collection, see Change Streams.

변경 스트림 $source를 사용하는 경우 소스 클러스터의 oplog 윈도우를 최소 24시간으로 구성하세요.

변경 스트림을 읽기 위해 Atlas Stream Processing은 oplog 컬렉션을 스캔합니다. 그 결과 로그에 COLLSCAN 경고가 표시될 수 있습니다. 이러한 경고는 정상적인 동작을 나타내며 오류 신호가 아닙니다.

If you configure config.fullDocument or config.fullDocumentBeforeChange to required, enable changeStreamPreAndPostImages on each collection before any write operations that you want to capture. If a post-image is unavailable for an event because you had not enabled the feature when the write occurred, or because the post-image expired, the stream processor fails. To learn how to enable pre- and post-images, see Change Streams with Document Pre- and Post-Images.

중요

You can resume a change stream only while the oplog collection still holds the operation that its resume token identifies. To ensure that your applications can resume their change streams, configure a Minimum Oplog Window longer than the longest interruption you expect.

Atlas 컬렉션 변경 스트림의 스트리밍 데이터로 작업하기 위해 $source 단계는 다음과 같은 프로토타입 형식을 갖습니다.

{
  "$source": {
    "connectionName": "<registered-connection>",
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "db" : "<source-db>",
    "coll" : ["<source-coll>",...],
    "initialSync": {
      "enable": <boolean>,
      "parallelism": <integer>
    },
    "readPreference": "<read-preference>",
    "readPreferenceTags": [
      {"<key>": "<value>"},
      . . .
    ],
    "config": {
      "startAfter": <start-token> | "startAtOperationTime": <timestamp>,
      "fullDocument": "<full-doc-condition>",
      "fullDocumentOnly": <boolean>,
      "fullDocumentBeforeChange": "<before-change-condition>",
      "pipeline": [{
        "<aggregation-stage>" : {
          <stage-input>,
          . . .
        },
        . . .
      }],
      "maxAwaitTimeMS": <time-ms>,
    }
  }
}

$source 단계에서는 다음 필드가 있는 문서를 사용합니다.

필드	유형	필요성	설명
`connectionName`	문자열	조건부	데이터를 수집할 연결 레지스트리 에서 연결을 식별하는 레이블입니다.
`timeField`	문서	옵션	수신 메시지에 대한 신뢰할 수 있는 타임스탬프를 정의하는 문서입니다. `timeField` 을(를) 사용하는 경우 다음 중 하나로 정의해야 합니다. 소스 메시지 필드를 인수로 사용하는 `$toDate` 표현식 소스 메시지 필드를 인수로 사용하는 `$dateFromString` 표현식입니다. `timeField` 을 선언하지 않으면 Atlas Stream Processing은 소스에서 제공한 메시지 타임스탬프에서 타임스탬프를 생성합니다.
`db`	문자열	필수 사항	`connectionName` 으로 지정된 Atlas 인스턴스에 호스팅된 MongoDB database의 이름입니다. 이 데이터베이스의 변경 스트림은 스트리밍 데이터 소스 역할을 합니다.
`coll`	문자열 또는 문자열 배열	필수 사항	Name of one or more MongoDB collections hosted on the Atlas instance specified by `connectionName`. The change stream of these collections act as the streaming data source. If you omit this field, your stream processor sources from a MongoDB Database Change Stream.
`initialSync`	문서	옵션	`initialSync` 구성 매개변수가 포함된 문서. Atlas Stream Processing `initialSync` allows you to ingest preexisting documents in an Atlas collection as though they were insert `changeEvent` documents. If you enable `initialSync`, when you start your stream processor, it first ingests and processes all existing documents in the collection before proceeding to ingest and process new incoming `changeEvent` documents. Once the `initialSync` is complete, it doesn't repeat. `coll`이 하나 이상의 컬렉션을 지정하는 경우 Atlas Stream Processing은 목록의 모든 컬렉션을 동기화합니다. 다중 컬렉션 동기화 순서, 계층 제한 및 예시에 대해 학습하려면, 다중 컬렉션 초기 동기화 구성을 참조하세요. Atlas Stream Processing은 스트림 프로세서가 시작될 때 동기화할 컬렉션 집합을 고정합니다. 그 점 이후에 생성된 컬렉션은 동기화되지 않습니다. 그리고 어떤 하나의 컬렉션 동기화가 실패하면 전체 스트림 프로세서가 실패합니다. `initialSync`를 활성화 하면 파이프라인 에서 `$hoppingWindow`, `$sessionWindow` 또는 `$tumblingWindow` 단계를 사용할 수 없습니다. 중요한: `initialSync` 고려 사항 및 제약 조건에 대한 제한 사항을 참조하세요.
`initialSync.enable`	부울	조건부	`initialSync`을(를) 활성화할 지 여부를 결정합니다. `initialSync` 필드를 선언하는 경우 이 필드를 설정해야 합니다.
`initialSync.parallelism`	integer	옵션	`initialSync` 작업을 프로세스 할 병렬 처리 수준을 결정합니다. 값을 지정하지 않으면 기본값은 `1`입니다. `coll`에서 여러 컬렉션을 지정하는 경우, 이 값은 각 컬렉션이 아니라 전체 컬렉션 목록에 적용됩니다. `initialSync` 지원되는 모든 `_id` 유형을 가진 컬렉션에 이 값을 적용합니다. Atlas Stream Processing이 예측 가능한 순서로 읽을 수 없는 유형도 포함됩니다. 설정할 수 있는 최대 값은 스트림 프로세서의 계층에 따라 다릅니다. 자세한 내용은 Atlas Stream Processing 계층 선택 가이드를 참조하십시오. 각 스트림 프로세서에는 해당 계층 에 따라 결정되는 최대 누적 병렬 처리 수 값이 있습니다. 스트림 프로세서의 누적 병렬 처리 수는 다음과 같이 계산됩니다. `parallelism total - parallelized stages` Where `parallelism total` is the sum of all `parallelism` values greater than 1 across the `$source`, $lookup, $merge, $emit, and $externalFunction stages, and `parallelized stages` is the number of these stages with `parallelism` values greater than `1`. 예시 를 들어 `$source` 단계에서 `parallelism` 값을 `4`로 설정하고, `$lookup` 단계에서 `parallelism` 값을 설정하지 않으며(따라서 기본값은 `1`), `$merge` 단계는 `parallelism` 값이 `2`인 경우 두 개의 `parallelized stages`이(가) 있고 스트림 프로세서의 누적 병렬 처리 수는 `(4 + 2) - 2`(으)로 계산됩니다. 스트림 프로세서가 해당 계층 의 최대 누적 병렬 처리 수를 초과하는 경우 Atlas Stream Processing 오류가 발생하고 의도한 병렬 처리 수준에 필요한 최소 프로세서 계층 알려줍니다. 오류를 해결하려면 프로세서를 더 높은 계층 으로 확장하다 하거나 단계의 병렬 처리 값을 줄여야 합니다. 자세한 학습은 Stream Processing을 참조하세요.
`readPreference`	문자열	옵션	Read preference for change stream and `initialSync` operations. 기본값은 `primary`입니다.
`readPreferenceTags`	배열	옵션	Read preference tags for change stream and `initialSync` operations.
`config`	문서	옵션	다양한 기본값을 재정의하는 필드가 포함된 문서입니다.
`config.startAfter`	token	조건부	The change event after which the source begins reporting. This takes the form of a resume token. `config.startAfter` 또는 `config.startAtOperationTime` 중 하나만 사용할 수 있습니다.
`config.startAtOperationTime`	타임스탬프 \| 날짜	조건부	소스가 보고를 시작해야 하는 optime입니다. `config.startAfter` 또는 `config.startAtOperationTime` 중 하나만 사용할 수 있습니다. Accepts MongoDB Extended JSON `$date` or `$timestamp` values.
`config.fullDocument`	문자열	조건부	change stream 소스에서 문서를 반환할지, 아니면 업데이트 발생 시 변경 사항만 반환할지를 제어하는 설정입니다. 다음 중 하나여야 합니다: `default` : `update` 작업에 대한 전체 문서 반환하지 않습니다. `updateLookup` : 업데이트 작업으로 인한 변경 사항 외에도 전체 문서 의 현재 다수 커밋 버전을 반환합니다. `required` : 전체 문서 반환해야 합니다. 전체 문서 사용할 수 없는 경우 스트림 프로세서가 실패합니다. 이는 전체 문서 사용할 수 없는 경우에도 삭제 이벤트를 발생시키는 삭제 작업에는 적용 되지 않습니다. `whenAvailable` : 사용 가능한 문서가 있을 때마다 전체 문서를 반환하고, 그렇지 않으면 변경 사항을 반환합니다. To use `required` or `whenAvailable` with a collection change stream, you must enable change stream Pre- and Post-Images on that collection.
`config.fullDocumentOnly`	부울	조건부	change stream 소스가 모든 메타데이터를 포함한 전체 이벤트 문서를 반환할지, 또는 `fullDocument` 의 내용만 반환할지를 제어하는 설정입니다. `true` 로 설정하면 소스는 `fullDocument` 의 내용만 반환합니다. If `fullDocument` is set to `required` or `whenAvailable`, you must enable change stream Pre- and Post-Images on that collection.
`config.fullDocumentBeforeChange`	문자열	옵션	change stream 소스가 출력에 원래 '변경 전' 상태의 문서를 포함할지 여부를 지정합니다. 다음 중 하나여야 합니다: `off` : `fullDocumentBeforeChange` 필드를 생략합니다. `required` : 변경 전 상태의 전체 문서를 반환해야 합니다. 변경 전 상태의 전체 문서를 사용할 수 없는 경우 스트림 프로세서가 실패합니다. `whenAvailable` : 전체 문서가 있는 경우 변경 전 상태로 전체 문서를 반환하고, 그렇지 않으면 `fullDocumentBeforeChange` 필드를 생략합니다. `fullDocumentBeforeChange` 에 값을 지정하지 않으면 기본값은 `off` 입니다. To use this field with a collection change stream, you must enable change stream Pre- and Post-Images on that collection.
`config.pipeline`	문서	옵션	Specifies an aggregation pipeline to filter change stream output before passing it on for further processing. This pipeline must conform to the parameters described in Modify Change Stream Output. 중요한: 각 변경 이벤트에는 `wallTime` 및 `clusterTime` 필드가 포함되어 있습니다. `$source` 이후의 Atlas Stream Processing 단계에서는 프로세서가 이러한 필드를 수집할 때 이러한 필드를 수신할 것으로 예상합니다. 변경 스트림 데이터가 올바르게 처리 되도록 하려면 `$source.config.pipeline`에서 이러한 필드를 수정하지 마세요.
`config.maxAwaitTimeMS`	integer	옵션	빈 배치를 반환하기 전에 변경 스트림 커서에 보고할 새 데이터 변경 내용을 기다리는 최대 시간(밀리초)입니다. 기본값은 `1000`입니다.

MongoDB database change stream

An Atlas database change stream allows applications to access real-time data changes on a single database. To learn how to open a change stream against a database, see Change Streams.

변경 스트림 $source를 사용하는 경우 소스 클러스터의 oplog 윈도우를 최소 24시간으로 구성하세요.

중요

Atlas 데이터베이스 변경 스트림의 스트리밍 데이터로 작업하기 위해 $source 단계는 다음과 같은 프로토타입 형식을 갖습니다.

{
  "$source": {
    "connectionName": "<registered-connection>",
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "db" : "<source-db>",
    "readPreference": "<read-preference>",
    "readPreferenceTags": [
      {"<key>": "<value>"},
      . . .
    ],
    "config": {
      "startAfter": <start-token> | "startAtOperationTime": <timestamp>,
      "fullDocument": "<full-doc-condition>",
      "fullDocumentOnly": <boolean>,
      "fullDocumentBeforeChange": "<before-change-condition>",
      "pipeline": [{
        "<aggregation-stage>" : {
          <stage-input>,
          . . .
        },
        . . .
      }]
    },
  }
}

$source 단계에서는 다음 필드가 있는 문서를 사용합니다.

필드	유형	필요성	설명
`connectionName`	문자열	조건부	데이터를 수집할 연결 레지스트리 에서 연결을 식별하는 레이블입니다.
`timeField`	문서	옵션	수신 메시지에 대한 신뢰할 수 있는 타임스탬프를 정의하는 문서입니다. `timeField` 을(를) 사용하는 경우 다음 중 하나로 정의해야 합니다. 소스 메시지 필드를 인수로 사용하는 `$toDate` 표현식 소스 메시지 필드를 인수로 사용하는 `$dateFromString` 표현식입니다. `timeField` 을 선언하지 않으면 Atlas Stream Processing은 소스에서 제공한 메시지 타임스탬프에서 타임스탬프를 생성합니다.
`db`	문자열	필수 사항	`connectionName` 으로 지정된 Atlas 인스턴스에 호스팅된 MongoDB database의 이름입니다. 이 데이터베이스의 변경 스트림은 스트리밍 데이터 소스 역할을 합니다.
`readPreference`	문자열	옵션	Read preference for change stream operations. 기본값은 `primary`입니다.
`readPreferenceTags`	배열	옵션	Read preference tags for change stream operations.
`config`	문서	옵션	다양한 기본값을 재정의하는 필드가 포함된 문서입니다.
`config.startAfter`	token	조건부	The change event after which the source begins reporting. This takes the form of a resume token. `config.startAfter` 또는 `config.startAtOperationTime` 중 하나만 사용할 수 있습니다.
`config.startAtOperationTime`	타임스탬프 \| 날짜	조건부	소스가 보고를 시작해야 하는 optime입니다. `config.startAfter` 또는 `config.startAtOperationTime` 중 하나만 사용할 수 있습니다. Accepts MongoDB Extended JSON `$date` or `$timestamp` values.
`config.fullDocument`	문자열	조건부	change stream 소스에서 문서를 반환할지, 아니면 업데이트 발생 시 변경 사항만 반환할지를 제어하는 설정입니다. 다음 중 하나여야 합니다: `default` : 서버 기본 동작을 사용합니다. : `update` 작업에 대한 전체 문서를 반환하지 않습니다. `updateLookup` : 업데이트 작업으로 인한 변경 사항 외에도 전체 문서 의 현재 다수 커밋 버전을 반환합니다. `required` : 전체 문서 반환해야 합니다. 전체 문서 사용할 수 없는 경우 스트림 프로세서가 실패합니다. 이는 전체 문서 사용할 수 없는 경우에도 삭제 이벤트를 발생시키는 삭제 작업에는 적용 되지 않습니다. `whenAvailable` : 사용 가능한 문서가 있을 때마다 전체 문서를 반환하고, 그렇지 않으면 변경 사항을 반환합니다. `fullDocument` 에 값을 지정하지 않으면 기본값은 `default` 입니다. To use `required` or `whenAvailable` with a database change stream, you must enable change stream Pre- and Post-Images on every collection in that database.
`config.fullDocumentOnly`	부울	조건부	change stream 소스가 모든 메타데이터를 포함한 전체 이벤트 문서를 반환할지, 또는 `fullDocument` 의 내용만 반환할지를 제어하는 설정입니다. `true` 로 설정하면 소스는 `fullDocument` 의 내용만 반환합니다. If `fullDocument` is set to `required` or `whenAvailable`, you must enable change stream Pre- and Post-Images on every collection in that database.
`config.fullDocumentBeforeChange`	문자열	옵션	change stream 소스가 출력에 원래 '변경 전' 상태의 문서를 포함할지 여부를 지정합니다. 다음 중 하나여야 합니다: `off` : `fullDocumentBeforeChange` 필드를 생략합니다. `required` : 변경 전 상태의 전체 문서를 반환해야 합니다. 변경 전 상태의 전체 문서를 사용할 수 없는 경우 스트림 프로세서가 실패합니다. `whenAvailable` : 전체 문서가 있는 경우 변경 전 상태로 전체 문서를 반환하고, 그렇지 않으면 `fullDocumentBeforeChange` 필드를 생략합니다. `fullDocumentBeforeChange` 에 값을 지정하지 않으면 기본값은 `off` 입니다. 이 필드를 데이터베이스 변경 스트림과 함께 사용하려면 해당 데이터베이스의 모든 컬렉션에서 변경 스트림 사전 및 사후 이미지 를 활성화해야 합니다.
`config.pipeline`	문서	옵션	Specifies an aggregation pipeline to filter change stream output at the point of origin. This pipeline must conform to the parameters described in Modify Change Stream Output. 중요한: 각 변경 이벤트에는 `wallTime` 및 `clusterTime` 필드가 포함되어 있습니다. `$source` 이후의 Atlas Stream Processing 단계에서는 프로세서가 이러한 필드를 수집할 때 이러한 필드를 수신할 것으로 예상합니다. 변경 스트림 데이터가 올바르게 처리 되도록 하려면 `$source.config.pipeline`에서 이러한 필드를 수정하지 마세요.
`config.maxAwaitTimeMS`	integer	옵션	빈 배치를 반환하기 전에 변경 스트림 커서에 보고할 새 데이터 변경 내용을 기다리는 최대 시간(밀리초)입니다. 기본값은 `1000`입니다.

MongoDB 클러스터 전체 변경 스트림 소스

변경 스트림 $source를 사용하는 경우 소스 클러스터의 oplog 윈도우를 최소 24시간으로 구성하세요.

중요

전체 Atlas 클러스터 변경 스트림에서 스트리밍 데이터를 처리하기 위해 $source 단계는 다음과 같은 프로토타입 형태를 갖습니다.

{
  "$source": {
    "connectionName": "<registered-connection>",
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "readPreference": "<read-preference>",
    "readPreferenceTags": [
      {"<key>": "<value>"},
      . . .
    ],
    "config": {
      "startAfter": <start-token> | "startAtOperationTime": <timestamp>,
      "fullDocument": "<full-doc-condition>",
      "fullDocumentOnly": <boolean>,
      "fullDocumentBeforeChange": "<before-change-condition>",
      "pipeline": [{
        "<aggregation-stage>" : {
          <stage-input>,
          . . .
        },
        . . .
      }]
    },
  }
}

$source 단계에서는 다음 필드가 있는 문서를 사용합니다.

필드	유형	필요성	설명
`connectionName`	문자열	조건부	데이터를 수집할 연결 레지스트리 에서 연결을 식별하는 레이블입니다.
`timeField`	문서	옵션	수신 메시지에 대한 신뢰할 수 있는 타임스탬프를 정의하는 문서입니다. `timeField` 을(를) 사용하는 경우 다음 중 하나로 정의해야 합니다. 소스 메시지 필드를 인수로 사용하는 `$toDate` 표현식 소스 메시지 필드를 인수로 사용하는 `$dateFromString` 표현식입니다. `timeField` 을 선언하지 않으면 Atlas Stream Processing은 소스에서 제공한 메시지 타임스탬프에서 타임스탬프를 생성합니다.
`readPreference`	문자열	옵션	Read preference for change stream operations. 기본값은 `primary`입니다.
`readPreferenceTags`	배열	옵션	Read preference tags for change stream operations.
`config`	문서	옵션	다양한 기본값을 재정의하는 필드가 포함된 문서입니다.
`config.startAfter`	token	조건부	The change event after which the source begins reporting. This takes the form of a resume token. `config.startAfter` 또는 `config.startAtOperationTime` 중 하나만 사용할 수 있습니다.
`config.startAtOperationTime`	날짜 \| 타임스탬프	조건부	소스가 보고를 시작해야 하는 optime입니다. `config.startAfter` 또는 `config.startAtOperationTime` 중 하나만 사용할 수 있습니다. Accepts MongoDB Extended JSON `$date` or `$timestamp` values.
`config.fullDocument`	문자열	조건부	change stream 소스에서 문서를 반환할지, 아니면 업데이트 발생 시 변경 사항만 반환할지를 제어하는 설정입니다. 다음 중 하나여야 합니다: `default` : 서버 기본 동작을 사용합니다. : `update` 작업에 대한 전체 문서를 반환하지 않습니다. `updateLookup` : 업데이트 작업으로 인한 변경 사항 외에도 전체 문서 의 현재 다수 커밋 버전을 반환합니다. `required` : 전체 문서 반환해야 합니다. 전체 문서 사용할 수 없는 경우 스트림 프로세서가 실패합니다. 이는 전체 문서 사용할 수 없는 경우에도 삭제 이벤트를 발생시키는 삭제 작업에는 적용 되지 않습니다. `whenAvailable` : 사용 가능한 문서가 있을 때마다 전체 문서를 반환하고, 그렇지 않으면 변경 사항을 반환합니다. `fullDocument` 에 값을 지정하지 않으면 기본값은 `default` 입니다. To use `required` or `whenAvailable` with a cluster change stream, you must enable change stream Pre- and Post-Images on every collection in that cluster.
`config.fullDocumentOnly`	부울	조건부	change stream 소스가 모든 메타데이터를 포함한 전체 이벤트 문서를 반환할지, 또는 `fullDocument` 의 내용만 반환할지를 제어하는 설정입니다. `true` 로 설정하면 소스는 `fullDocument` 의 내용만 반환합니다. If `fullDocument` is set to `required` or `whenAvailable`, you must enable change stream Pre- and Post-Images on every collection in that cluster.
`config.fullDocumentBeforeChange`	문자열	옵션	change stream 소스가 출력에 원래 '변경 전' 상태의 문서를 포함할지 여부를 지정합니다. 다음 중 하나여야 합니다: `off` : `fullDocumentBeforeChange` 필드를 생략합니다. `required` : 변경 전 상태의 전체 문서를 반환해야 합니다. 변경 전 상태의 전체 문서를 사용할 수 없는 경우 스트림 프로세서가 실패합니다. `whenAvailable` : 전체 문서가 있는 경우 변경 전 상태로 전체 문서를 반환하고, 그렇지 않으면 `fullDocumentBeforeChange` 필드를 생략합니다. `fullDocumentBeforeChange` 에 값을 지정하지 않으면 기본값은 `off` 입니다. 이 필드를 데이터베이스 변경 스트림과 함께 사용하려면 해당 데이터베이스의 모든 컬렉션에서 변경 스트림 사전 및 사후 이미지 를 활성화해야 합니다.
`config.pipeline`	문서	옵션	Specifies an aggregation pipeline to filter change stream output at the point of origin. This pipeline must conform to the parameters described in Modify Change Stream Output. Atlas Stream Processing 수집된 각 변경 이벤트에서 `wallTime` 및 `clusterTime` 필드를 수신할 것으로 예상합니다. 변경 스트림 데이터를 올바르게 처리 하려면 `$source.config.pipeline`에서 이러한 필드를 수정하지 마세요.
`config.maxAwaitTimeMS`	integer	옵션	빈 배치를 반환하기 전에 변경 스트림 커서에 보고할 새 데이터 변경 내용을 기다리는 최대 시간(밀리초)입니다. 기본값은 `1000`입니다.

AWS Kinesis Data Stream

Atlas Stream Processing AWS Kinesis 스트림에 대한 Private Link 연결 생성을 지원합니다. 자세한 학습 은 Kinesis Private Link 연결 추가를 참조하세요.

AWS Kinesis 데이터 스트림의 데이터로 작업하기 위해 $source 단계의 프로토타입 형식은 다음과 같습니다.

{
  "$source": {
    "connectionName": "<registered-connection>",
    "stream": "<stream-name>",
    "region": "<aws-region>",
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "tsFieldName": "<field-name>",
    "shardIdleTimeout": {
      "size": <duration-number>,
      "unit": "<duration-unit>"
    },
    "config": {
      "consumerARN": "<aws-arn>",
      "initialPosition": <initial-position>,
      reshardDetectionIntervalSecs: <interval>
    }
  }
}

$source 단계에서는 다음 필드가 있는 문서를 사용합니다.

필드	유형	필요성	설명
`connectionName`	문자열	필수 사항	연결 레지스트리에서 데이터를 수집할 연결을 식별하는 레이블입니다.
`config.consumerARN`	문자열	필수 사항	Kinesis 소비자에 해당하는ARN입니다. 소비자는 향상된 팬아웃을 사용해야 합니다.
`stream`	문자열	필수 사항	메시지를 스트림 할AWS Kinesis 데이터 스트림 입니다.
`region`	문자열	조건부	지정된스트림 있는 AWS 리전 입니다. Kinesis 서로 다른 리전에서 동일한 이름을 가진 여러 데이터 스트림을 지원합니다. 동일한 연결 내 둘 이상의 리전에서 데이터 스트림에 동일한 이름을 사용하는 경우 이 필드 사용하여 사용할 이름과 리전 의 조합을 지정해야 합니다.
`timeField`	문서	옵션	수신 메시지에 대한 신뢰할 수 있는 타임스탬프를 정의하는 문서입니다. `timeField` 을(를) 사용하는 경우 다음 중 하나로 정의해야 합니다. 소스 메시지 필드를 인수로 사용하는 `$toDate` 표현식 소스 메시지 필드를 인수로 사용하는 `$dateFromString` 표현식입니다. `timeField` 을 선언하지 않으면 Atlas Stream Processing은 소스에서 제공한 메시지 타임스탬프에서 타임스탬프를 생성합니다.
`tsFieldName`	문자열	옵션	프로젝션된 문서의 타임스탬프에 대한 필드 이름입니다. 이 필드를 사용하여 기본 타임스탬프 필드 이름을 재정의합니다.
`shardIdleTimeout`	문서	옵션	워터마크 계산에서 샤드가 무시되기 전에 샤드 유휴 상태로 허용되는 시간을 지정하는 문서입니다. 이 필드 기본값 으로 비활성화되어 있습니다. 유휴 상태로 인해 앞으로 이동하지 않는 샤드를 처리하다 하려면 이 필드 의 값을 설정하다 .
`shardIdleTimeout.size`	문서	옵션	샤드 유휴 시간 초과 기간을 지정하는 숫자입니다.
`shardIdleTimeout.unit`	문서	옵션	샤드 유휴 시간 초과 기간의 시간 단위입니다. `unit` 의 값은 다음 중 하나일 수 있습니다. `"ms"` (밀리초) `"second"` `"minute"` `"hour"` `"day"`
`config`	문서	옵션	다양한 기본값을 재정의하는 필드가 포함된 문서입니다.
`config.initialPosition`	문자열	옵션	메시지 수집을 시작할 Kinesis 데이터 스트림 기록의 위치입니다. 다음 중 하나여야 합니다. `"TRIM_HORIZON"`: 샤드 에서 가장 오래된 메시지부터 수집을 시작합니다. `"LATEST"`: 샤드 의 최신 메시지 에서 수집 을 시작 합니다 . `"AT_TIMESTAMP"`: 특정 타임스탬프에서 수집을 시작합니다. `config.atTimestamp`이(가) 필요합니다. `"AT_SEQUENCE_NUMBER"`: 특정 시퀀스 번호에서 수집을 시작합니다. `"AFTER_SEQUENCE_NUMBER"`: 특정 시퀀스 번호 이후 수집 시작 기본값은 `"LATEST"`입니다.
`config.atTimestamp`	날짜	조건부	메시지 수집을 시작할 시간입니다. `config.initialPosition` 이 `"AT_TIMESTAMP"`인 경우 필수입니다.
`config.reshardDetectionIntervalSecs`	integer	옵션	리샤딩을 위해Kinesis 스트림 통한 데이터 흐름 속도에 대한 검사 간격(초)입니다. `300`초로 기본 설정됩니다.

문서 배열

문서 배열에서 작업하기 위해 $source 단계의 프로토타입 형식은 다음과 같습니다.

{
  "$source": {
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "documents" : [{source-doc},...] | <expression>
  }
}

$source 단계에서는 다음 필드가 있는 문서를 사용합니다.

필드	유형	필요성	설명
`timeField`	문서	옵션	수신 메시지에 대한 신뢰할 수 있는 타임스탬프를 정의하는 문서입니다. `timeField` 을(를) 사용하는 경우 다음 중 하나로 정의해야 합니다. 소스 메시지 필드를 인수로 사용하는 `$toDate` 표현식 소스 메시지 필드를 인수로 사용하는 `$dateFromString` 표현식입니다. `timeField` 을 선언하지 않으면 Atlas Stream Processing은 소스에서 제공한 메시지 타임스탬프에서 타임스탬프를 생성합니다.
`documents`	배열	조건부	Array of documents to use as a streaming data source. 이 필드의 값은 객체 배열이거나 객체 배열로 평가되는 표현식일 수 있습니다. `connectionName` 필드를 사용할 때는 이 필드를 사용하지 마세요.

크론 일정

To generate a document on a recurring schedule instead of reading from a connection, the $source stage has the following prototype form:

{
  "$source": {
    "schedule": "<cron-expression>",
    "tsFieldName": "<timestamp-field-name>"
  }
}

$source 단계에서는 다음 필드가 있는 문서를 사용합니다.

필드	유형	필요성	설명
`schedule`	문자열	필수 사항	Six-field cron expression that determines when the stage emits a document. The fields are, in order: second, minute, hour, day of month, month, and day of week. Atlas Stream Processing evaluates the expression in UTC. 각 필드 모든 값, 단일 값, 범위 (예: `1-5`), 단계(예: `0/15`) 또는 쉼표로 구분된 목록(예: `1,15,30`)에 대해 `*`을 허용합니다. 월에는 `JAN`부터 `DEC`까지, 요일에는 `SUN`부터 `SAT`까지 이름을 사용할 수 있습니다. `CRON_TZ` 접두사를 포함하지 마세요. To learn about how Atlas Stream Processing resolves an expression that restricts both the day of month and the day of week, see Behavior.
`tsFieldName`	문자열	옵션	단계에서 예약된 타임스탬프를 프로젝션하는 필드 의 이름입니다. 기본값은 `_ts`입니다.

행동

$source는 표시되는 모든 파이프라인의 첫 번째 단계여야 합니다. 파이프라인당 $source 단계는 하나만 사용할 수 있습니다.

Kafka $source 단계의 경우 Atlas Stream Processing 소스 주제 내의 여러 파티션에서 병렬로 읽습니다. 파티션 제한은 프로세서 계층 에 따라 결정됩니다. 자세한 학습은 Stream Processing 청구 참조를참조하세요.

cron $source 단계의 경우, 각 예정된 시간에 예정된 타임스탬프가 포함된 단일의 빈 문서가 생성됩니다. 파이프라인의 후반 단계를 사용하여 문서를 채우십시오.

When a schedule expression restricts both the day of month and the day of week, Atlas Stream Processing emits a document only at the scheduled times that satisfy both fields. This differs from cron implementations that emit a document when either field matches.

cron $source 단계를 가진 스트림 프로세서가 다시 시작되면, 실행되지 않는 동안 노친 예정된 시간에 대해 각각 문서를 발급합니다. 중지한 후 놓친 예약된 시간에 대한 문서를 생성하지 않습니다.

A stream processor with a cron $source stage runs continuously between scheduled times, and Atlas bills it at your processor tier for that entire period. An infrequent schedule does not reduce the cost of running the processor. To learn more, see the Stream Processing billing reference.

예시

Kafka 예시

A streaming data source generates detailed weather reports from various locations, conformant to the schema of the Sample Weather Dataset. The following aggregation has three stages:

$source 단계는 my_weatherdata이라는 주제 에서 이러한 보고서를 수집하는 Apache Kafka 브로커와의 연결을 설정하여 수집되는 각 기록 후속 집계 단계에 노출합니다. 또한 이 단계에서는 프로젝션하는 타임스탬프 필드 의 이름을 재정의하여 ingestionTime로 설정합니다.
The $match stage excludes documents that have a dewPoint.value of less than or equal to 5.0 and passes along the documents with dewPoint.value greater than 5.0 to the next stage.
The $merge stage writes the output to an Atlas collection named stream in the sample_weatherstream database. If no such database or collection exist, Atlas creates them.

[{
  "$source": {
    "connectionName": "sample_weatherdata",
    "topic": "my_weatherdata"
  }
},
{
  "$match": { "dewPoint.value": { "$gt": 5 } }
},
{
  "$merge": {
    "into": {
      "connectionName": "weatherStreamOutput",
      "db": "sample_weatherstream",
      "coll": "stream"
    }
  }
}]

결과 sample_weatherstream.stream 컬렉션의 문서를 보려면 Atlas 클러스터에 연결하고 다음 명령을 실행하세요.

db.getSiblingDB("sample_weatherstream").stream.find()

{
  _id: ObjectId('66ad2edfd4fcac13b1a28ce3'),
  airTemperature: { quality: '1', value: 27.7 },
  atmosphericPressureChange: {
    quantity24Hours: { quality: '9', value: 99.9 },
    quantity3Hours: { quality: '1' },
    tendency: { code: '1', quality: '1' }
  },
  atmosphericPressureObservation: {
    altimeterSetting: { quality: '1', value: 1015.9 },
    stationPressure: { quality: '1', value: 1021.9 }
  },
  callLetters: 'CGDS',
  dataSource: '4',
  dewPoint: { quality: '9', value: 25.7 },
  elevation: 9999,
  extremeAirTemperature: {
    code: 'N',
    period: 99.9,
    quantity: '9',
    value: -30.4
  },
  ingestionTime: ISODate('2024-08-02T19:09:18.071Z'),
  liquidPrecipitation: {
    condition: '9',
    depth: 160,
    period: 24,
    quality: '2'
  },
  pastWeatherObservationManual: {
    atmosphericCondition: { quality: '1', value: '8' },
    period: { quality: '9', value: 3 }
  },
  position: { coordinates: [ 153.3, 50.7 ], type: 'Point' },
  precipitationEstimatedObservation: {
    discrepancy: '4',
    estimatedWaterDepth: 4
  },
  presentWeatherObservationManual: { condition: '53', quality: '1' },
  pressure: { quality: '1', value: 1016.3 },
  qualityControlProcess: 'V020',
  seaSurfaceTemperature: { quality: '9', value: 27.6 },
  sections: [ 'AA2', 'SA1', 'MW1', 'AG1', 'GF1' ],
  skyCondition: {
    cavok: 'N',
    ceilingHeight: { determination: 'C', quality: '1', value: 6900 }
  },
  skyConditionObservation: {
    highCloudGenus: { quality: '1', value: '05' },
    lowCloudGenus: { quality: '9', value: '03' },
    lowestCloudBaseHeight: { quality: '9', value: 150 },
    lowestCloudCoverage: { quality: '1', value: '05' },
    midCloudGenus: { quality: '9', value: '08' },
    totalCoverage: { opaque: '99', quality: '1', value: '06' }
  },
  skyCoverLayer: {
    baseHeight: { quality: '9', value: 99999 },
    cloudType: { quality: '9', value: '05' },
    coverage: { quality: '1', value: '04' }
  },
  st: 'x+35700-027900',
  type: 'SAO',
  visibility: {
    distance: { quality: '1', value: 4000 },
    variability: { quality: '1', value: 'N' }
  },
  waveMeasurement: {
    method: 'I',
    seaState: { code: '99', quality: '9' },
    waves: { height: 99.9, period: 14, quality: '9' }
  },
  wind: {
    direction: { angle: 280, quality: '9' },
    speed: { quality: '1', rate: 30.3 },
    type: '9'
  }
}

참고

위 사례는 대표적인 예시입니다. 스트리밍 데이터는 정적이지 않으며 각 사용자는 서로 다른 문서를 보게 됩니다.

변경 스트림 예시

다음 집계는 샘플 데이터세트가 로드된 Atlas 클러스터에 연결되는 cluster0-collection 소스에서 데이터를 수집합니다. 스트림 처리 작업 공간을 생성하고 연결 레지스트리에 Atlas 클러스터에 대한 연결을 추가하는 방법을 알아보려면 Atlas Stream Processing 시작하기를 참조하세요. 이 집계는 두 단계를 실행하여 변경 스트림을 열고 sample_weatherdata 데이터베이스의 data 컬렉션에 대한 변경 사항을 기록합니다.

The $source stage connects to the cluster0-collection source and opens a change stream against the data collection in the sample_weatherdata database.
The $merge stage writes the filtered change stream documents to an Atlas collection named data_changes in the sample_weatherdata database. If no such collection exists, Atlas creates it.

[{
  "$source": {
    "connectionName": "cluster0-connection",
    "db": "sample_weatherdata",
    "coll": "data"
  }
},
{
  "$merge": {
    "into": {
      "connectionName": "cluster0-connection",
      "db": "sample_weatherdata",
      "coll": "data_changes"
    }
  }
}]

The following mongosh command deletes a data document:

db.getSiblingDB("sample_weatherdata").data.deleteOne(
   { _id: ObjectId("5553a99ae4b02cf715120e4b") }
)

After the data document is deleted, the stream processor writes the change stream event document to the sample_weatherdata.data_changes collection. To view the documents in the resulting sample_weatherdata.data_changes collection, use mongosh to connect to your Atlas cluster and run the following command:

db.getSiblingDB("sample_weatherdata").data_changes.find()

[
  {
    _id: {
      _data: '8267A3D7A3000000012B042C0100296E5A1004800951B8EDE4430AB5C1B254BB3C96D6463C6F7065726174696F6E54797065003C64656C6574650046646F63756D656E744B65790046645F696400645553A99AE4B02CF715120E4B000004'
    },
    clusterTime: Timestamp({ t: 1738790819, i: 1 }),
    documentKey: { _id: ObjectId('5553a99ae4b02cf715120e4b') },
    ns: { db: 'sample_weatherdata', coll: 'data' },
    operationType: 'delete',
    wallTime: ISODate('2025-02-05T21:26:59.313Z')
  }
]

데이터베이스 변경 스트림 예시

The following aggregation ingests data from the db-change-stream-connection source, which connects to an Atlas cluster loaded with the Sample Mflix Dataset Collections dataset. To learn how to create a stream processing workspace and add a connection to an Atlas cluster to the connection registry, see Get Started with Atlas Stream Processing. This aggregation runs two stages to open a change stream against the sample_mflix source database and record changes to the db_changes collection in the sample_mflix_changes sink database:

The $source stage connects to the db-change-stream-connection source and opens a change stream against the sample_mflix source database. The config.startAtOperationTime field sets the time after which the source begins reporting. This example sets the startAt value to start one minute in the past.
The $merge stage writes the change stream documents to an Atlas collection named db_changes in the sample_mflix_changes sink database.

const startAt = new Date(Date.now() - 60 * 1000);
const pipeline = [
   {
      $source: {
         connectionName: "db-change-stream-connection",
         db: "sample_mflix",
         config: {
            startAtOperationTime: startAt
         }
      }
   },
   {
      $merge: {
         into: {
            connectionName: "db-change-stream-connection",
            db: "sample_mflix_changes",
            coll: "db_changes"
         }
      }
   }
];

소스 클러스터에서 다음 명령을 실행하여 스트림 프로세서의 동작을 확인합니다. 연결하려면, mongosh 를 통해 클러스터에 연결을 참조하세요.

sample_mflix 소스 데이터베이스에서 movies 컬렉션과 comments 컬렉션에 문서를 삽입합니다:

db.getSiblingDB("sample_mflix").movies.insertOne({
   title: "The Stream Processor",
   year: 2026
})
db.getSiblingDB("sample_mflix").comments.insertOne({
   name: "Ada Lovelace",
   text: "A fine film about data in motion."
})

문서를 삽입하면, 스트림 프로세서는 각 삽입마다 변경 스트림 이벤트 문서를 sample_mflix_changes.db_changes 컬렉션에 쓰기 (write)합니다. 결과로 생성된 sample_mflix_changes.db_changes 컬렉션에서 문서를 확인하기 위해 클러스터에서 다음 명령을 실행하세요:

db.getSiblingDB("sample_mflix_changes").db_changes.find(
   {},
   { _id: 0, clusterTime: 1, ns: 1, operationType: 1, fullDocument: 1 }
)

[
  {
    clusterTime: Timestamp({ t: 1787778766, i: 1 }),
    fullDocument: {
      _id: ObjectId('6a8f56ce73c5d4d0ee569a0d'),
      title: 'The Stream Processor',
      year: 2026
    },
    ns: { db: 'sample_mflix', coll: 'movies' },
    operationType: 'insert'
  },
  {
    clusterTime: Timestamp({ t: 1787778771, i: 1 }),
    fullDocument: {
      _id: ObjectId('6a8f56d373c5d4d0ee569a0e'),
      name: 'Ada Lovelace',
      text: 'A fine film about data in motion.'
    },
    ns: { db: 'sample_mflix', coll: 'comments' },
    operationType: 'insert'
  }
]

각 이벤트는 타임스탬프, 전체 문서 변경, operationType 필드 및 ns 필드를 포함합니다. ns 필드는 원본 데이터베이스와 컬렉션의 이름을 표시합니다. 따라서 각 변경 사항을 생성한 컬렉션을 구분할 수 있습니다.

클러스터 전체 변경 스트림 예시

The following aggregation ingests data from the cluster-changestream-connection source, which connects to an Atlas cluster loaded with the Sample Mflix Dataset Collections dataset, and writes to the cluster-changestream-sink-connection destination, which connects to a separate Atlas cluster. To learn how to create a stream processing workspace and add a connection to an Atlas cluster to the connection registry, see Get Started with Atlas Stream Processing. This aggregation runs two stages to open a cluster-wide change stream on the source cluster and record changes to the events collection in the cluster_changes database on the destination cluster. This $source stage omits db and coll, so it reports changes from every database and collection on the source cluster, rather than from a single collection or database.

The $source stage connects to the cluster-changestream-connection source and opens a change stream against the entire source cluster. The config.startAtOperationTime field specifies that the stream processor should begin reporting changes that occur on or after a specified time.
The $merge stage writes the change stream documents to an Atlas collection named events in the cluster_changes database on the destination cluster. If no such database or collection exists, Atlas creates them.

[{
  "$source": {
    "connectionName": "cluster-changestream-connection",
    "config": {
      "startAtOperationTime": {"$date": "2024-08-19T18:00:00.000Z"}
    }
  }
},
{
  "$merge": {
    "into": {
      "connectionName": "cluster-changestream-sink-connection",
      "db": "cluster_changes",
      "coll": "events"
    }
  }
}]

The following mongosh command inserts a document into the movies collection in the sample_mflix database on the source cluster:

db.getSiblingDB("sample_mflix").movies.insertOne(
   { _id: ObjectId("66c1a1f1f1f1f1f1f1f1f1f1"), title: "Example Movie" }
)

문서가 삽입된 후, 스트림 프로세서는 변경 스트림 이벤트 문서를 대상 클러스터의 cluster_changes.events 컬렉션에 쓰기 (write)합니다.

결과 cluster_changes.events 컬렉션 의 문서를 보려면 대상 클러스터 에 대해 다음 명령을 실행 . 연결하려면 mongosh 통해 클러스터에 연결을 참조하세요.

db.getSiblingDB("cluster_changes").events.find()

[
  {
    _id: {
      _data: '826A8F5E65000000012B042C0100296E5A10049CC3BB749CB641F4BF5A032A2ABE2FB1463C6F7065726174696F6E54797065003C696E736572740046646F63756D656E744B65790046645F6964006466C1A1F1F1F1F1F1F1F1F1F1000004'
    },
    clusterTime: Timestamp({ t: 1787780709, i: 1 }),
    documentKey: { _id: ObjectId('66c1a1f1f1f1f1f1f1f1f1f1') },
    fullDocument: { _id: ObjectId('66c1a1f1f1f1f1f1f1f1f1f1'), title: 'Example Movie' },
    ns: { db: 'sample_mflix', coll: 'movies' },
    operationType: 'insert',
    wallTime: ISODate('2026-08-26T21:45:09.059Z')
  }
]

출력 문서 의 ns 필드 변경 사항이 소스 클러스터 의 sample_mflix.movies 삽입에서 비롯된 것임을 보여줍니다. 변경 스트림 집계 파이프라인 대상 클러스터 의 cluster_changes.events 컬렉션 에 대한 이러한 변경 사항을 반영합니다.

문서 배열 예시

The following aggregation uses an inline document array as the streaming data source, containing weather observations for three locations. The array uses the same schema as the Sample Weather Dataset. This aggregation runs three stages:

The $source stage defines an inline documents array of weather observations as the streaming data source, and uses timeField to designate the timestamp field of each document as the authoritative timestamp.
The $match stage passes along only the documents with dewPoint.value greater than 5.0 to the next stage.
The $merge stage writes the output to an Atlas collection named stream in the sample_weatherstream database. If no such database or collection exist, Atlas creates them.

[{
  "$source": {
    "documents": [
      {
        "location": "New York",
        "timestamp": ISODate('2024-01-15T08:00:00Z'),
        "temp": 23.5,
        "dewPoint": { "value": 6.2 }
      },
      {
        "location": "Los Angeles",
        "timestamp": ISODate('2024-01-15T08:05:00Z'),
        "temp": 18.2,
        "dewPoint": { "value": 4.8 }
      },
      {
        "location": "Chicago",
        "timestamp": ISODate('2024-01-15T08:10:00Z'),
        "temp": 26.8,
        "dewPoint": { "value": 7.5 }
      }
    ]
  }
},
{
  "$match": { "dewPoint.value": { "$gt": 5.0 } }
},
{
  "$merge": {
    "into": {
      "connectionName": "weatherStreamOutput",
      "db": "sample_weatherstream",
      "coll": "stream"
    }
  }
}]

결과 sample_weatherstream.stream 컬렉션의 문서를 보려면 Atlas 클러스터에 연결하고 다음 명령을 실행하세요.

db.getSiblingDB("sample_weatherstream").stream.find()

[
  {
    _id: ObjectId('67a6c3df14fcac13b1a21a01'),
    dewPoint: { value: 6.2 },
    location: 'New York',
    temp: 23.5,
    timestamp: ISODate('2024-01-15T08:00:00.000Z')
  },
  {
    _id: ObjectId('67a6c3df14fcac13b1a21a03'),
    dewPoint: { value: 7.5 },
    location: 'Chicago',
    temp: 26.8,
    timestamp: ISODate('2024-01-15T08:10:00.000Z')
  }
]

Solar Stream 예시

다음 집계는 샘플 스트리밍 데이터 소스 sample_stream_solar에서 보고서를 필터링합니다. 결과를 solar-cluster-connection 연결에 보관합니다. 스트림 처리 작업 공간을 생성하고 Atlas Stream Processing 작업 공간에 연결을 추가하는 방법을 알아보려면 Atlas Stream Processing 시작하기 및 Atlas Stream Processing 연결 추가를 참조하세요. 이 집계는 sample_stream_solar 소스에서 보고서를 필터링하고 solarDb 데이터베이스의 solarColl 컬렉션에 결과를 쓰기 (write) 위해 세 단계를 실행합니다:

The $source stage connects to the sample_stream_solar source. The timeField field converts the timestamp field of each incoming report to a date using $dateFromString.
The $match stage excludes documents where device_id is device_8, and passes along reports from all other devices to the next stage.
The $merge stage writes the output to the solarColl collection in the solarDb database of the solar-cluster-connection cluster.

[{
  "$source": {
    "connectionName": "sample_stream_solar",
    "timeField": {
      "$dateFromString": { "dateString": "$timestamp" }
    }
  }
},
{
  "$match": { "device_id": { "$ne": "device_8" } }
},
{
  "$merge": {
    "into": {
      "connectionName": "solar-cluster-connection",
      "db": "solarDb",
      "coll": "solarColl"
    }
  }
}]

참고

sample_stream_solar 소스는 매초마다 샘플 문서를 생성하는 테스트 전용 연결이며 래피드 프로토타입 제작을 위한 것입니다.

solar-cluster-connection 클러스터에서 다음 명령을 실행하여 스트림 프로세서의 동작을 확인합니다. 연결하려면 mongosh 를 통해 클러스터에 연결을 참조하세요.

결과 solarDb.solarColl 컬렉션의 문서를 보려면 실행하세요.

db.getSiblingDB("solarDb").solarColl.find()

[
  {
    _id: ObjectId('6a8f5f0352c6ffe64c8884ad'),
    device_id: 'device_3',
    event_type: 0,
    group_id: 8,
    max_watts: 450,
    obs: { watts: 428, temp: 10 },
    timestamp: '2024-08-12T21:47:47.287+00:00'
  }
]

Each document includes the device_id, event_type, group_id, max_watts, obs, and timestamp fields of a solar device report. You can observe that no document in the output has a device_id of device_8, because the $match stage excludes those reports before the $merge stage writes the remaining documents to solarColl.

참고

위 사례는 대표적인 예시입니다. 스트리밍 데이터는 정적이지 않으며 각 사용자는 서로 다른 문서를 보게 됩니다.

크론 일정 예시

다음 집계는 연결에서 읽는 대신 5분 마다 문서를 생성합니다. 이 집계는 세 단계로 구성됩니다.

The $source stage emits an empty document at second 0 of every fifth minute, in UTC.
The $project stage labels each document with a job name and copies the scheduled timestamp from the _ts field into a runAt field.
The $merge stage writes the output to an Atlas collection named heartbeats in the sample_weatherstream database. If no such database or collection exist, Atlas creates them.

[{
  "$source": {
    "schedule": "0 0/5 * * * *"
  }
},
{
  "$project": {
    "job": "five-minute-heartbeat",
    "runAt": "$_ts"
  }
},
{
  "$merge": {
    "into": {
      "connectionName": "weatherStreamOutput",
      "db": "sample_weatherstream",
      "coll": "heartbeats"
    }
  }
}]

결과 sample_weatherstream.heartbeats 컬렉션의 문서를 보려면 Atlas 클러스터에 연결하고 다음 명령을 실행하세요.

db.getSiblingDB("sample_weatherstream").heartbeats.find()

[
  {
    _id: ObjectId('68b8a1c014fcac13b1a21b01'),
    job: 'five-minute-heartbeat',
    runAt: ISODate('2026-09-03T14:00:00.000Z')
  },
  {
    _id: ObjectId('68b8a2f414fcac13b1a21b02'),
    job: 'five-minute-heartbeat',
    runAt: ISODate('2026-09-03T14:05:00.000Z')
  },
  {
    _id: ObjectId('68b8a42814fcac13b1a21b03'),
    job: 'five-minute-heartbeat',
    runAt: ISODate('2026-09-03T14:10:00.000Z')
  }
]

참고

위 사례는 대표적인 예시입니다. 표시되는 타임스탬프는 스트림 프로세서를 시작하는 시점에 따라 달라집니다.

돌아가기

집계 단계

$validate

$source 단계(Stream Processing)

정의

구문

Apache Kafka 브로커

MongoDB collection change stream

MongoDB database change stream

MongoDB 클러스터 전체 변경 스트림 소스

AWS Kinesis Data Stream

문서 배열

크론 일정

행동

예시

Kafka 예시

변경 스트림 예시

데이터베이스 변경 스트림 예시

클러스터 전체 변경 스트림 예시

문서 배열 예시

Solar Stream 예시

크론 일정 예시

`$source` 단계(Stream Processing)