/ /

`$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`	string	常に	ソースとして使用される接続のタイプ。
`stream.source.ts`	ISODate	常に	取り込み時点でのレコードの日時。
`stream.source.topic`	string	条件付き	ストリームがレコードを取り込む 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`	string	条件付き	Atlas Stream Processing が指定されたドキュメントに対して実行しようとしたデータベース操作の種類。Atlas 変更ストリームソースにのみ適用されます。
`stream.source.ns`	ドキュメント	条件付き	Atlas Stream Processing がドキュメントを取得する名前空間を含むドキュメント。Atlas 変更ストリームソースにのみ適用されます。
`stream.source.ns.db`	string	条件付き	Atlas Stream Processing が操作を試行するデータベース名。Atlas 変更ストリームソースにのみ適用されます。この値は、コレクション変更ストリームまたはデータベース変更ストリームソースのすべてのドキュメントで同じです。クラスター変更ストリームソースの場合は異なります。
`stream.source.ns.coll`	string	条件付き	Atlas Stream Processing が操作を試行するコレクションの名前。Atlas 変更ストリームソースにのみ適用されます。この値は、コレクション変更ストリームソースのすべてのドキュメントで同じです。データベース変更ストリームまたはクラスター変更ストリームソースの場合は異なります。
`stream.source.documentKey._id`	ドキュメント	条件付き	ソースドキュメントのオブジェクト ID を含むドキュメント。Atlas 変更ストリームソースにのみ適用されます。
`stream.source.initialSync.phase`	string	条件付き	最初の同期作業の現在の状態。最初の同期中の Atlas 変更ストリームソースにのみ適用されます。
`stream.source.kinesisStream`	string	条件付き	Atlas Stream Processing がドキュメントを取得する Kinesis Data Stream の名前。AWS Kinesis ソースにのみ適用されます。
`stream.source.shardId`	string	条件付き	Atlas Stream Processing がドキュメントを取得する Kinesis Data Stream 内のシャードの ID。AWS Kinesis ソースにのみ適用されます。
`stream.source.sequenceNumber`	string	条件付き	Kinesis Data Stream から取得したドキュメントの一意の識別子。Amazon Web Services Kinesis ソースにのみ適用されます。
`stream.source.partitionKey`	string	条件付き	ソースドキュメントが属するパーティションの一意の識別子。AWS Kinesis ソースにのみ適用されます。
`stream.window`	ドキュメント	条件付き	ウィンドウのメタデータを含むドキュメント。ドキュメントがウィンドウで処理された場合にのみ適用されます。
`stream.window.start`	ISODate	条件付き	ウィンドウ開始時間。ドキュメントがウィンドウで処理された場合にのみ適用されます。
`stream.window.end`	ISODate	条件付き	ウィンドウの閉じる時間。ドキュメントがウィンドウで処理された場合にのみ適用されます。
`stream.window.partition`	string	条件付き	ドキュメントが属すするセッションウィンドウパーティション。ドキュメントがセッションウィンドウで処理された場合にのみ適用されます。
`stream.https`	ドキュメント	条件付き	$https ステージのメタデータを含むドキュメント。プロセシングの失敗が `$https` ステージで発生した場合にのみ適用されます。
`stream.https.url`	string	条件付き	`$https` ステージのターゲット URL。プロセシングの失敗が `$https` ステージで発生した場合にのみ適用されます。
`stream.https.method`	string	条件付き	`$https` ステージで使用される HTTP リクエストメソッド。プロセシングの失敗が `$https` ステージで発生した場合にのみ適用されます。
`stream.https.httpStatusCode`	整数	条件付き	リクエストの HTTP 応答ステータスコード。プロセシングの失敗が `$https` ステージで発生した場合にのみ適用されます。
`stream.https.responseTimeMs`	整数	条件付き	リクエストの応答時間（ミリ秒）。プロセシングの失敗が `$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` 集計ステージ（ストリームプロセシング）