/ /

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

정의

$meta 표현식은 문서의 모든 스트리밍 메타데이터를 포함하는 객체를 반환합니다. 이 데이터를 전체 스트림 또는 다음 Atlas Stream Processing 집계 단계 중 하나에 대해 노출할 수 있습니다.

$meta 표현식의 프로토타입 형식은 다음과 같습니다.

{ "$meta": <string> }

"source": {
  "type": "<source-type>",
  "ts": {
    "$date": "<datetime>"
  },
  "topic": "<string>",
  "partition": <int>,
  "offset": <int>,
  "key": "<kafka-key>",
  "headers": [
    {
      "k": "<header-key>",
      "v": "<header-value>"
    }
  ],
  "operationType": "<db-operation>",
  "ns": {
    "db": "<namespace-db>",
    "coll": "<namespace-coll>"
  },
  "documentKey": {
    "_id": {
      "$oid": "<object-id>"
    }
  },
  "initialSync": {
    "phase": "<sync-state>"
  }
  "kinesisStream": "<kinesis-name>",
  "shardId": "<kinesis-shard-id>",
  "sequenceNumber": "<doc-uuid>",
  "partitionKey": "<partition-id>",
}
"window": {
  "start": <ISODate>,
  "end": <ISODate>,
  "partition": "<session-partition>"
},
"https": {
  "url": "<target-url>",
  "method": "<request-method>",
  "httpStatusCode": <http-code>,
  "responseTimeMs": <response-time-ms>
}

구문

$meta 표현식은 메타데이터 소스의 정규화된 점 구문 경로에 해당하는 단일 문자열 입력을 사용합니다. 이 경로의 루트는 "stream"이어야 합니다. 다음 경로를 쿼리할 수 있습니다.

경로	유형	조건부	설명
`stream`	객체	항상	파이프라인에 구성된 `$source` 단계 및 모든 창 단계 또는 `$https` 단계에 대한 모든 메타데이터입니다.
`stream.source`	문서	항상	`$source` 단계의 메타데이터가 포함된 문서입니다.
`stream.source.type`	문자열	항상	소스로 사용되는 연결의 유형입니다.
`stream.source.ts`	ISODate	항상	수집 시점에 기록된 날짜와 시간입니다.
`stream.source.topic`	문자열	조건부	스트림이 기록을 수집하는 Kafka 주제입니다. Kafka 소스에만 적용됩니다.
`stream.source.partition`	integer	조건부	스트림이 기록을 수집하는 Kafka 주제의 파티션입니다. Kafka 소스에만 적용됩니다.
`stream.source.offset`	integer	조건부	Kafka 소스 파티션 내에서 메시지 순서 및 큐 위치를 추적하는 오프셋입니다. Kafka 소스에만 적용됩니다.
`stream.source.key`	string\|int\|long\|double\|object\|binData	조건부	분할 및 부하 분산을 위해 Kafka 메시지에 할당된 키입니다. Kafka 소스에만 적용됩니다.
`stream.source.headers`	배열	조건부	Kafka 메시지 메타데이터를 설명하는 키-값 쌍의 집합입니다. Kafka 소스에만 적용됩니다.
`stream.source.operationType`	문자열	조건부	Atlas Stream Processing이 주어진 문서에 대해 수행하려고 시도한 데이터베이스 작업 유형입니다. Atlas 변경 스트림 소스에만 적용됩니다.
`stream.source.ns`	문서	조건부	Atlas Stream Processing이 문서를 가져오는 네임스페이스가 포함된 문서입니다. Atlas 변경 스트림 소스에만 적용됩니다.
`stream.source.ns.db`	문자열	조건부	Atlas Stream Processing이 작업을 시도하는 데이터베이스의 이름. Atlas 변경 스트림 소스에만 적용됩니다. 이 값은 컬렉션 변경 스트림 또는 데이터베이스 변경 스트림 소스의 모든 문서에 대해 동일합니다. 클러스터 변경 스트림 소스에 대해 다릅니다.
`stream.source.ns.coll`	문자열	조건부	Atlas Stream Processing이 작업을 시도하는 컬렉션의 이름입니다. Atlas 변경 스트림 소스에만 적용됩니다. 이 값은 컬렉션 변경 스트림 소스의 모든 문서에 대해 동일합니다. 데이터베이스 변경 스트림 또는 클러스터 변경 스트림 소스에 대해 다릅니다.
`stream.source.documentKey._id`	문서	조건부	소스 문서의 객체 ID가 포함된 문서입니다. Atlas 변경 스트림 소스에만 적용됩니다.
`stream.source.initialSync.phase`	문자열	조건부	초기 동기화 작업의 현재 상태. 초기 동기화 동안 Atlas 변경 스트림 소스에만 적용됩니다.
`stream.source.kinesisStream`	문자열	조건부	Atlas Stream Processing이 문서를 가져오는 Kinesis Data Stream 의 이름입니다. Amazon Web Services Kinesis 소스에만 적용됩니다.
`stream.source.shardId`	문자열	조건부	Atlas Stream Processing 문서를 소싱하는 Kinesis Data Stream 내 샤드 의 ID . Amazon Web Services Kinesis 소스에만 적용됩니다.
`stream.source.sequenceNumber`	문자열	조건부	Kinesis Data Stream에서 가져온 문서의 고유 식별자입니다. Amazon Web Services Kinesis 소스에만 적용됩니다.
`stream.source.partitionKey`	문자열	조건부	소스 문서가 속한 파티션의 고유 식별자입니다. Amazon Web Services Kinesis 소스에만 적용됩니다.
`stream.window`	문서	조건부	창 메타데이터가 포함된 문서입니다. 문서가 창에서 처리된 경우에만 적용됩니다.
`stream.window.start`	ISODate	조건부	창 열림 시간입니다. 문서가 창에서 처리된 경우에만 적용됩니다.
`stream.window.end`	ISODate	조건부	창 닫기 시간. 문서가 창에서 처리된 경우에만 적용됩니다.
`stream.window.partition`	문자열	조건부	문서가 속한 세션 창 파티션. 문서가 세션 창에서 처리된 경우에만 적용됩니다.
`stream.https`	문서	조건부	$https 단계에 대한 메타데이터가 포함된 문서입니다. `$https` 단계에서 처리 실패가 발생한 경우에만 적용됩니다.
`stream.https.url`	문자열	조건부	`$https` 단계의 타겟 URL. 처리 실패가 `$https` 단계에서 발생한 경우에만 적용됩니다.
`stream.https.method`	문자열	조건부	`$https` 단계에서 사용되는 HTTP 요청 메서드입니다. 처리 실패가 `$https` 단계에서 발생한 경우에만 적용됩니다.
`stream.https.httpStatusCode`	int	조건부	요청의 HTTP 응답 상태 코드입니다. 처리 실패가 `$https` 단계에서 발생한 경우에만 적용됩니다.
`stream.https.responseTimeMs`	int	조건부	요청의 응답 시간(밀리초). 처리 실패가 `$https` 단계에서 발생한 경우에만 적용됩니다.

행동

Atlas Stream Processing $meta 표현식 기존 MongoDB $meta 집계 표현식의 모든 기능을 제공합니다. 그러나 표준 MongoDB 집계 쿼리 에서는 $meta 의 Atlas Stream Processing 버전에 특정한 기능을 사용할 수 없습니다.

예시

다음 예시는 데이터가 수집된 Kafka 소스 주제의 배열로 스트림의 출력을 강화합니다.

{
  $source: {
    connectionName: "kafka",
    topic: ["t1", "t2", "t3"]
  }
},
{
  $emit: {
    connectionName: "kafka",
    topic: {
      $concat: [
        {
          $meta: "stream.source.topic"
        },
        "out"
      ]
    }
  }
}

다음 예시는 각 창의 시작 시간을 보고하는 스트림에 필드를 추가합니다.

{
  $source: {
    connectionName: "kafka",
    topic: "t1"
  }
},
{
  $hoppingWindow: . . .
},
{
  $addFields: {
    start: { $meta: "stream.window.start" }
  }
}

돌아가기

$currentDate

$createUUID

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

정의

구문

행동

예시

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