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

정의

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

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

{ "$meta": <string> }

{
  source: {
    type: string,
    ts: date,
    source.topic: string
    source.partition: int
    source.offset: int
    source.key: string|int|long|double|object|binData
    source.headers: array[obj]
  },
  window: {
    start: date,
    end: date
  },
  https: {
    url: string
    method: string
    httpStatusCode: int
    responseTimeMs: int
  }
}

구문

$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 메시지 메타데이터를 설명하는 키-값 쌍의 집합입니다.
`stream.window`	객체	`$hoppingWindow` 또는 `$tumblingWindow` 단계의 모든 메타데이터입니다. 이 객체는 문서가 `$hoppingWindow` 또는 `$tumblingWindow` 단계에 들어갈 때만 설정됩니다.
`stream.window.start`	ISODate	현재 창의 시작 시간입니다.
`stream.window.end`	ISODate	현재 창의 종료 시간입니다.
`stream.https`	객체	`$https` 단계의 모든 메타데이터입니다. 이 객체는 `$https` 단계가 문서를 출력할 때만 설정됩니다.
`stream.https.url`	문자열	`$https` 단계에서 데이터를 소싱하는 URL입니다.
`stream.https.method`	문자열	URL로 전송된 HTTPS 요청 메서드입니다.
`stream.https.httpStatusCode`	int	URL로 전송된 요청에 대한 HTTP 응답 코드입니다.
`stream.https.responseTimeMs`	int	URL에서 응답을 받는 데 걸린 시간(밀리초)입니다.

행동

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

예시

다음 예시는 데이터가 수집된 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