/ /

`$emit` 集計ステージ（ストリームプロセシング）

定義

$emit ステージでは、接続レジストリでメッセージを出力する接続を指定します。次の接続タイプがサポートされています。

Apache Kafkaエージェント
MongoDB 時系列コレクション
AWS Kinesisデータストリーム
Amazon Web Services S3バケット
Azure BLOB ストレージ
Google Cloud パブリック/サブスクライブ

配置

$emit は、表示されるすべてのパイプラインの最後のステージである必要があります。パイプラインごとに使用できる$emitステージは 1 つだけです。

構文

Apache Kafka ブロック

処理されたデータをApache Kafkaプロバイダーに書き込むには、次のプロトタイプ形式で $emitパイプラインステージを使用します。

{
  "$emit": {
    "connectionName": "<registered-connection>",
    "topic": "<target-topic>" | <expression>,
    "schemaRegistry": {
      "connectionName": "<schema-registry-name>",
      "valueSchema": {
        type: "<schema-type>",
        schema: <schema-name>,
        options: {
          subjectNameStrategy: "<topic-name-strategy>",
          autoRegisterSchemas: true
        }
      }
    },
    "config": {
      "acks": <number-of-acknowledgements>,
      "compression_type": "<compression-type>",
      "dateFormat": "default" | "ISO8601",
      "headers": "<expression>",
      "key": "<key-string>" | { key-document },
      "keyFormat": "<serialization-type>",
      "outputFormat": "basicJson" | "canonicalJson" | "relaxedJson",
      "tombstoneWhen": <expression>
    }
  }
}

$emitステージは、次のフィールドを持つドキュメントを取得します。

フィールド

タイプ

必要性

説明

connectionName

string

必須

データを取り込む接続の名前（接続レジストリに表示されます）。

topic

string または式

必須

Apache Kafka の名前メッセージを送信するトピック。

schemaRegistry

ドキュメント

任意

平均直列化されたソースへの書き込みをサポートするためにスキーマレジストリの使用を有効にするドキュメント。

この機能を有効にするには、スキーマ Registry 接続を作成する必要があります。

schemaRegistry.connectionName

string

条件付き

Avro 非直列化に使用するスキーマレジストリ接続の名前。

schemaRegistry.valueSchema

ドキュメント|式

条件付き

直列化スキーマのプロパティを定義するドキュメント、またはそれを評価する式。

schemaRegistry.valueSchema.type

string

条件付き

スキーマレジストリを使用する直列化のタイプ。Atlas Stream Processing は現在、スキーマレジストリ接続を介した "avro"直列化をサポートしています。

schemaRegistry.valueSchema.schema

ドキュメント

条件付き

スキーマ宣言を定義するドキュメント。

schemaRegistry.valueSchema.options

ドキュメント

任意

スキーマレジストリ接続のオプションの構成パラメーターを定義するドキュメント。

schemaRegistry.valueSchema.options.autoRegisterSchemas

ブール値

任意

認識されないスキーマを持つドキュメントをプロセシングする際に、スキーマを自動的に登録するかどうかを判定するトグル。false に設定されている場合、認識されないスキーマを持つドキュメントはデッドレターキュー (DLQ)に送信されます。

デフォルトは true です。

schemaRegistry.valueSchema.options.subjectNameStrategy

string

条件付き

自動登録されるスキーマのサブジェクト名を決定する方法。次のいずれかである必要があります。

"TopicNameStrategy": サブジェクト名としてKafka {topic} 名を使用します。
"RecordNameStrategy": サブジェクト名として、Avroレコード名を使用します。
"TopicRecordNameStrategy": Kafka {topic} 名と Avroレコード名を組み合わせたものをサブジェクト名として使用します。

デフォルトは "TopicNameStrategy" です。このパラメータは、schemaRegistry.valueSchema.options.autoRegisterSchemas が true に設定されている場合にのみ設定できます。

config

ドキュメント

任意

のさまざまなデフォルト値を上書きするフィールドを含むドキュメント。

config.acks

string

任意

操作を成功させるためにApache Kafkaクラスターから必要な確認応答の数。$emit

デフォルト値は "all" です。Atlas Stream Processing は以下の値をサポートしています。

"-1"
"0"
"1"
"all"

config.compression_type

string

任意

制作者が生成するすべてのデータの圧縮形式。デフォルトは「なし」になっています（つまり、圧縮は行われません）。有効な値は次のとおりです。

none
gzip
snappy
lz4
zstd

圧縮はデータの完全なバッチに使用されるため、バッチ処理の効率は圧縮率に影響します。バッチ処理が多いほど、圧縮率が向上します。

config.dateFormat

string

任意

日付値のための日付フォーマット。有効な値は次のとおりです。

default outputFormat のデフォルトを使用する。
ISO8601 - ミリ秒単位の精度（YYYY-MM-DDTHH:mm:ss.sssZ）を含む ISO8601 形式で、日付を文字列に変換する。

以下に例を挙げます。

次の入力をご覧ください。

{ "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" }

config.headers

式

任意

出力メッセージに追加するヘッダー。式は、オブジェクトまたは配列のいずれかに評価される必要があります。

式がオブジェクトと評価される場合、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

config.key

オブジェクト | string | 式

任意

Apache Kafka として評価される式メッセージキー。

config.keyを指定する場合は、 config.keyFormatを指定する必要があります。

config.keyFormat

string

条件付き

Apache Kafkaキーデータを直列化するために使用されるデータ型。次のいずれかの値である必要があります。

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

デフォルトは binData です。config.key を指定する場合は、config.keyFormat を指定する必要があります。ドキュメントの config.key が指定されたデータ型に正常に直列化されない場合、Atlas Stream Processing ではデッドレターキュー (DLQ) に送信されます。

config.outputFormat

string

任意

Apache Kafka にメッセージを送信するときに使用する JSON 形式。次のいずれかの値である必要があります。

"basicJson"
"canonicalJson"
"relaxedJson"

デフォルトは "relaxedJson" です。

詳しくは、「基本的なJSON 」を参照してください。

config.tombstoneWhen

式

任意

Kafkaに null を発行するタイミングを決定する式。式はブール値true または false のいずれかに評価する必要があります。式が特定のドキュメントに対して true と評価されると、Atlas Stream Processing はその場所で null をKafkaシンクに出力します。式がfalse と評価された場合、Atlas Stream Processing は $emit ステージに達したときにドキュメントが存在するように出力します。

式がブール値値に評価できない場合、または評価できない場合、Atlas Stream Processing はドキュメントをDLQ に書込みます。

この設定を使用すると、$emit.config.key と $emit.config.keyFormat の値を指定する場合にトピック圧縮を有効にできます。これらの値を指定しない場合、この式がtrue と評価されたときに Atlas Stream Processing は引き続き null を発行しますが、これらによってKafkaトピックの圧縮はトリガーされません。

Atlas 時系列コレクション

処理されたデータを Atlas 時系列コレクションに書き込むには、次のプロトタイプ形式で$emitパイプラインステージを使用します。

{
  "$emit": {
    "connectionName": "<registered-connection>",
    "db": "<target-db>" | <expression>,
    "coll": "<target-coll>" | <expression>,
    "parallelism": <number>,
    "timeseries": {
      <options>
    }
  }
}

$emitステージは、次のフィールドを持つドキュメントを取得します。

フィールド	タイプ	必要性	説明
`connectionName`	string	必須	データを取り込む接続の名前（接続レジストリに表示されます）。
`db`	string \|式	必須	ターゲット時系列コレクションを含む Atlas データベースに解決される、または式の名前。
`coll`	string \|式	必須	書込み先の Atlas時系列コレクションに解決される、または式の名前。
`parallelism`	integer	任意	時系列シンク演算子が時系列コレクションにデータを書込むために使用する内部書込みスレッドの数。これにより、書込み操作が複数のスレッドに分散され、大容量のワークロードのスループットが向上します。デフォルトは `1` です。設定できる最大値は、ストリームプロセッサーの階層によって異なります。詳しくは、Atlas Stream Processing 階層選択ガイドを参照してください。各ストリームプロセッサには、その階層によって決定される最大累積並列処理値があります。ストリームプロセッサの累積並列処理は、次のように計算されます。 `parallelism total - parallelized stages` は、、$lookup、$merge、$emit、$externalFunction `$source`ステージにおける `parallelism total`を超えるすべての`parallelism` 値の合計であり、1`parallelized stages` はこれらのステージの数で{ を持つものを含む`parallelism` より大きい`1` の値。例、`$source` ステージが `4` の `parallelism` 値を設定し、`$lookup` ステージでは `parallelism` 値が設定されていない（デフォルトは `1`）、かつ `$merge` ステージが `parallelism` に設定されている場合: `2` の } 値がある場合は 2 つの `parallelized stages` があり、ストリームプロセッサの累積並列処理は `(4 + 2) - 2` として計算されます。ストリームプロセッサがその階層の最大累積並列処理を超える場合、Atlas Stream Processing はエラーをスローし、目的のレベルの並列処理に必要な最小プロセッサ階層について提案します。エラーを解決するには、プロセッサをより高い階層に増やすアップするか、ステージの並列処理値を低くする必要があります。詳しくは、Stream Processingをご覧ください。
`timeseries`	ドキュメント	任意	コレクションの時系列フィールドを定義するドキュメント。

注意

時系列コレクション内のドキュメントの最大サイズは 4 MB です。「時系列コレクションの制限」を参照してください。

重要

動的式の要件

db または coll フィールドに式を使用する場合は、timeseries 構成で timeField オプションを指定する必要があります。これにより、ターゲットコレクションが存在しない場合でも、Atlas Stream Processing は時系列コレクションを適切に作成できるようになります。

以下に例を挙げます。

{
  "$emit": {
    "connectionName": "atlas1",
    "db": "$targetDatabase",
    "coll": "$targetCollection",
    "timeseries": {
      "timeField": "timestamp"
    }
  }
}

AWS Kinesis

Atlas Stream Processing は、 AWS Kinesisストリームへの Private Link 接続の作成をサポートしています。詳細については、Kinesis Private Link 接続の追加を参照してください。

処理されたデータをAWS Kinesisに書き込むには、次のプロトタイプ形式で $emitパイプラインステージを使用します。

{
  "$emit": {
    "connectionName": "<registered-connection>",
    "stream": "<stream-name>",
    "region": "<aws-region>",
    "partitionKey": "<key>" | <field> | <expression>
    "config": {
      "outputFormat": "basicJson" | "canonicalJson" | "relaxedJson",
      "dateFormat": "default" | "ISO8601",
    }
  }
}

$emitステージは、次のフィールドを持つドキュメントを取得します。

フィールド

タイプ

必要性

説明

connectionName

string

必須

データを取り込む接続の名前（接続レジストリに表示されます）。

stream

string

必須

接続するKinesisデータストリームの名前。

region

string

任意

Kinesis Data Stream が動作するリージョン。AWSは同じ名前の複数のストリームをサポートしており、それぞれが異なるリージョンにあります。このパラメーターにより、Atlas Stream Processing はこのようなストリームを区別することができます。

partitionKey

string |式

必須

各レコードを受信する Kinesis シャードを決定するために使用されるキー。リテラルの string または string に評価される式である必要があります。

config

ドキュメント

任意

のさまざまなデフォルト値を上書きするフィールドを含むドキュメント。

config.outputFormat

string

任意

Kinesisにメッセージを送信するときに使用するJSON形式。次のいずれかの値である必要があります。

"basicJson"
"canonicalJson"
"relaxedJson"

デフォルトは "relaxedJson" です。

詳しくは、「基本的なJSON 」を参照してください。

config.dateFormat

string

任意

日付値のための日付フォーマット。有効な値は次のとおりです。

default outputFormat のデフォルトを使用する。
ISO8601 - ミリ秒単位の精度（YYYY-MM-DDTHH:mm:ss.sssZ）を含む ISO8601 形式で、日付を文字列に変換する。

以下に例を挙げます。

次の入力をご覧ください。

{ "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" }

AWS S3

To write processed data to an AWS S3 bucket sink connection, use the $emit pipeline stage with the following prototype form:

{
  "$emit": {
    "connectionName": "<registered-connection>",
    "bucket": "<target-bucket>" | <expression>,
    "region": "<target-region>",
    "path": "<key-prefix>" | <expression>,
    "parallelism": <number>,
    "partitionBy": <expression>,
    "config": {
      "writeOptions": {
        "count": <doc-count>,
        "bytes": <threshold>,
        "interval": {
          "size": <unit-count>,
          "unit": "<time-denomination>"
        }
      },
      "delimiter": "<delimiter>",
      "outputFormat": "basicJson" | "canonicalJson" | "relaxedJson",
      "dateFormat": "default" | "ISO8601",
      "compression": "gzip" | "snappy",
      "compressionLevel": <level>
    }
  }
}

$emitステージは、次のフィールドを持つドキュメントを取得します。

フィールド

タイプ

必要性

説明

connectionName

string

必須

接続レジストリに表示される、データを書き込む先の接続の名前。

bucket

string |式

必須

Name of the S3 bucket to which to write data, specified as a literal string or an expression that evaluates to a string.

region

string

任意

対象バケットが存在する Amazon Web Services リージョンの名前です。Stream Processing ワークスペースを Amazon Web Services リージョンでホストしている場合、このパラメータはそのリージョンがデフォルトになります。そうでない場合は、Stream Processing ワークスペースのホストリージョンに最も近い AWS リージョンがデフォルトになります。

path

string |式

必須

S3バケットに書き込まれたオブジェクトのキーのプレフィックス。リテラルのプレフィックス string または、string を評価する式である必要があります。

parallelism

integer

任意

S3 Sink 演算子が S3 にデータを書込むために使用する内部書込みスレッドの数。これにより、書込み操作が複数のスレッドに分散され、大容量のワークロードのスループットが向上します。

デフォルトは 1 です。

設定できる最大値は、ストリームプロセッサーの階層によって異なります。詳しくは、Atlas Stream Processing 階層選択ガイドを参照してください。

各ストリームプロセッサには、その階層によって決定される最大累積並列処理値があります。ストリームプロセッサの累積並列処理は、次のように計算されます。

parallelism total - parallelized stages

は、、$lookup、$merge、$emit、$externalFunction $sourceステージにおける parallelism totalを超えるすべてのparallelism 値の合計であり、1parallelized stages はこれらのステージの数で{ を持つものを含むparallelism より大きい1 の値。

例、$source ステージが 4 の parallelism 値を設定し、$lookup ステージでは parallelism 値が設定されていない（デフォルトは 1）、かつ $merge ステージが parallelism に設定されている場合: 2 の } 値がある場合は 2 つの parallelized stages があり、ストリームプロセッサの累積並列処理は (4 + 2) - 2 として計算されます。

ストリームプロセッサがその階層の最大累積並列処理を超える場合、Atlas Stream Processing はエラーをスローし、目的のレベルの並列処理に必要な最小プロセッサ階層について提案します。エラーを解決するには、プロセッサをより高い階層に増やすアップするか、ステージの並列処理値を低くする必要があります。詳しくは、Stream Processingをご覧ください。

partitionBy

式

任意

parallelism が 1 より大きい場合にドキュメントを書込みスレッド間で分割する方法を決定する式。評価された式の式によって、各ドキュメントを処理するスレッドが決まります。

parallelism が 1 より大きいが、partitionBy が指定されていない場合は、指定されていないパーティショニング戦略が使用されます。

config

ドキュメント

任意

さまざまなデフォルト値を上書きする追加のパラメーターを含むドキュメント。

config.writeOptions

ドキュメント

任意

書き込み動作を制御する追加パラメーターを含むドキュメントです。これらのパラメーターは、どの閾値が最初に満たされるかに応じて書き込み動作をトリガーします。

For example, if the ingested documents reach the config.writeOptions.count threshold without reaching the config.writeOptions.interval threshold, the stream processor still emits these documents to S3 according to the config.writeOptions.count threshold.

config.writeOptions.count

integer

任意

Number of documents to group into each file written to S3.

config.writeOptions.bytes

integer

任意

ファイルがS3 に書き込まれる前に蓄積する必要がある最小バイト数を指定します。バイト数は、最終出力ファイルのサイズではなく、パイプラインによって取り込まれたBSONドキュメントのサイズによって決まります。

デフォルトは 32 MB です。

config.writeOptions.interval

ドキュメント

任意

ドキュメントを一括書き込みするためのタイマーをsizeとunitの組み合わせとして指定します。

デフォルトは 1 分です。いかなるunitに対してもsizeを0に設定することはできません。最大間隔は7日です。

config.writeOptions.interval.size

integer

条件付き

writeOptions.interval.unit で指定されたユニット数。その後、ストリームプロセッサは S3 にドキュメントを一括書込みます。

デフォルトは1です。0 の size は設定できません。writeOptions.interval を定義する場合は、このパラメーターも必ず定義する必要があります。

config.writeOptions.interval.unit

string

条件付き

一括書き込みタイマーをカウントする時間の単位です。このパラメーターは以下の値をサポートしています：

ms
second
minute
hour
day

デフォルトはminuteです。writeOptions.intervalを定義する場合は、このパラメータも定義しなければなりません。

config.delimiter

string

任意

出力されたファイル内の各エントリの間の区切り文字。

デフォルトは \n です。

config.outputFormat

string

任意

Specifies the output format of the JSON written to S3. Must be one of the following values:

"basicJson"
"canonicalJson"
"relaxedJson"

デフォルトは「relaxedJson」です。

詳しくは、「基本JSONと拡張JSON 」を参照してください。

config.dateFormat

string

任意

日付値のための日付フォーマット。有効な値は次のとおりです。

default outputFormat のデフォルトを使用する。
ISO8601 - ミリ秒単位の精度（YYYY-MM-DDTHH:mm:ss.sssZ）を含む ISO8601 形式で、日付を文字列に変換する。

例として、次のレコードをパイプラインに追加した場合:

{ "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" }

config.compression

string

任意

使用する圧縮アルゴリズムの名称。次のいずれかの値でなければなりません：

"none"
"gzip"
"snappy"

config.compressionLevel

integer

条件付き

出力されるメッセージに適用する圧縮レベルです。1-9 を含む値をサポートしており、値が大きいほど圧縮率が高くなります。

デフォルトは 6 です。

このパラメータは、config.compression が gzip に設定されている場合にのみ有効です。それを他の圧縮タイプで設定するとエラーが返されます。

Azure BLOB ストレージ

処理されたデータをAzure Blob ストレージに書き込むには、次のプロトタイプ形式で $emitパイプラインステージを使用します。

{
  "$emit": {
    "connectionName": "<registered-connection>",
    "container": "<storage-container>",
    "path": "<prefix-key>" | <expression>,
    "parallelism": <int>,
    "partitionBy": "<key>" | <expression>,
    "config": {
      "dateFormat": "default" | "ISO8601",
      "compression": "none" | "gzip" | "snappy",
      "compressionLevel": [1-9],
      "outputFormat": "relaxedJson" | "canonicalJson" | "basicJson",
      "delimiter": "<delimiter>",
      "writeOptions": {
        "count": <int>,
        "interval": {
          "size": <int>,
          "unit": "ms" | "second" | "minute" | "hour" | "day"
        },
        "bytes": <int>
      }
    }
  }
}

$emitステージは、次のフィールドを持つドキュメントを取得します。

フィールド

タイプ

必要性

説明

connectionName

string

必須

データを書き込む接続の名前（接続レジストリに表示）。

container

string

必須

ブロブを書き込むストレージコンテナ。

path

string |式

必須

Azure Blob Storage に書き込まれたオブジェクトのキーのプレフィックス。リテラル string または string を評価する式である必要があります。

config

ドキュメント

任意

さまざまなデフォルト値を上書きする追加のパラメーターを含むドキュメント。

config.dateFormat

string

任意

日付値のための日付フォーマット。有効な値は次のとおりです。

default outputFormat のデフォルトを使用する。
ISO8601 - ミリ秒単位の精度（YYYY-MM-DDTHH:mm:ss.sssZ）を含む ISO8601 形式で、日付を文字列に変換する。

例として、次のレコードをパイプラインに追加した場合:

{ "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" }

config.compression

string

任意

使用する圧縮アルゴリズムの名称。次のいずれかの値でなければなりません：

"none"
"gzip"
"snappy"

このフィールドに値を指定しない場合、Atlas Stream Processing は圧縮なしでデータを書込みます。

config.compressionLevel

integer

条件付き

出力されるメッセージに適用する圧縮レベルです。1-9 を含む値をサポートしており、値が大きいほど圧縮率が高くなります。

デフォルトは 6 です。

このパラメータは、config.compression が gzip に設定されている場合にのみ有効です。それを他の圧縮タイプで設定するとエラーが返されます。

config.outputFormat

string

任意

Azure Blob ストレージに書き込まれる JSON の出力形式を指定します。次のいずれかの値である必要があります。

"basicJson"
"canonicalJson"
"relaxedJson"

デフォルトは "relaxedJson" です。

詳しくは、「基本JSONと拡張JSON 」を参照してください。

config.delimiter

string

任意

出力されたファイル内の各エントリの間の区切り文字。

デフォルトは \n です。

config.writeOptions

ドキュメント

任意

例、取り込まれたドキュメントが config.writeOptions.interval のしきい値に達せずに config.writeOptions.count のしきい値に達した場合、ストリームプロセッサは引き続き config.writeOptions.count しきい値に従ってこれらのドキュメントをAzure Blob Storage に出力します。

config.writeOptions.count

integer

任意

Azure Blob Storage に書き込まれた各ファイルにグループ化するドキュメントの数。

config.writeOptions.bytes

integer

任意

ファイルがAzure Blob ストレージに書き込まれる前に蓄積する必要がある最小バイト数を指定します。バイト数は、最終出力ファイルのサイズではなく、パイプラインによって取り込まれたBSONドキュメントのサイズによって決まります。

デフォルトは 32 MB です。

config.writeOptions.interval

ドキュメント

任意

ドキュメントを一括書き込みするためのタイマーをsizeとunitの組み合わせとして指定します。

デフォルトは 1 分です。いかなるunitに対してもsizeを0に設定することはできません。最大間隔は7日です。

config.writeOptions.interval.size

integer

条件付き

writeOptions.interval.unit で指定され、その後ストリームプロセッサがAzure Blob ストレージにドキュメントを一括書込みするユニット数。

デフォルトは1です。0 の size は設定できません。writeOptions.interval を定義する場合は、このパラメーターも必ず定義する必要があります。

config.writeOptions.interval.unit

string

条件付き

一括書き込みタイマーをカウントする時間の単位です。このパラメーターは以下の値をサポートしています：

ms
second
minute
hour
day

デフォルトはminuteです。writeOptions.intervalを定義する場合は、このパラメータも定義しなければなりません。

parallelism

integer

任意

Atlas Stream Processing が Azure Blob ストレージにデータを書き込むために使用する内部書込みスレッドの数。これにより、書込み操作が複数のスレッドに分散され、大容量のワークロードのスループットが向上します。

デフォルトは 1 です。最大値は 16 です。

partitionBy

string |式

任意

parallelism が 1 より大きい場合に Atlas Stream Processing が書込みスレッド全体でドキュメントを分割する方法を決定する文字列または式。評価された式の式によって、各ドキュメントを処理するスレッドが決まります。

parallelism が 1 より大きいが、partitionBy が指定されていない場合、Atlas Stream Processing はラウンドロビン戦略を使用してドキュメントをスレッドに割り当てます。

Google Cloud パブリック/サブスクライブ

処理されたデータを Google Cloud Pub/Sub に書き込むには、次のプロトタイプ形式の $emitパイプラインステージを使用します。

{
  "$emit": {
    "connectionName": "<registered-connection>",
    "topic": "<topic>" | <expression>,
    "projectId": "<project-id>",
    "region": "<pubsub-region>",
    "orderingKey": "<key>" | <expression>,
    "attributes": {
      "<key1>": "<value1>" | <expression>,
      . . .
      "<keyN>": "<valueN>" | <expression>
    },
    "config": {
      "dateFormat": "default" | "ISO8601",
      "outputFormat": "relaxedJson"| "canonicalJson" | "basicJson",
    }
  }
}

$emitステージは、次のフィールドを持つドキュメントを取得します。

フィールド

タイプ

必要性

説明

connectionName

string

必須

データを書き込む接続の名前（接続レジストリに表示）。

projectId

string

必須

公開するプロジェクトのID 。

topic

string |式

必須

メッセージを公開する特定のプロジェクト内のトピックの名前。string、または string を評価する式である必要があります。

region

string

必須

リクエストの送信先となるロケーションエンドポイント。利用可能なリージョンについては、 Pub/Sub のドキュメントを参照してください。

orderingKey

string |式

任意

メッセージの公開を順序付ける公開データ内のキー。If you set an orderingKey, Atlas Stream Processing は、サブスクライブシンクが、ソースが公開する順序でそのキーを含むすべての公開メッセージを受信するようにします。

このフィールドに string 値を使用して、静的 orderingKey を設定します。動的な orderingKey には string として評価される式を使用します。フィールドパスに対して具体的に評価する式は、orderingKey をドキュメント内のフィールドの値に設定します。

attributes

ドキュメント

任意

メッセージに追加する属性を定義するドキュメント。属性はキーと値のペアの形式をとります。ここでは、

各キーは、Atlas Stream Processing がメッセージに追加するキーを決定する string またはフィールドパス式です。
各値は、string または string として評価されるフィールドパス式のいずれかです。

config

ドキュメント

任意

さまざまなデフォルト値を上書きする追加のパラメーターを含むドキュメント。

config.dateFormat

string

任意

日付値のための日付フォーマット。有効な値は次のとおりです。

default outputFormat のデフォルトを使用する。
ISO8601 - ミリ秒単位の精度（YYYY-MM-DDTHH:mm:ss.sssZ）を含む ISO8601 形式で、日付を文字列に変換する。

例として、次のレコードをパイプラインに追加した場合:

{ "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" }

config.outputFormat

string

任意

Google Cloud Pub/Sub に書き込まれるJSONの出力形式を指定します。次のいずれかの値である必要があります。

"basicJson"
"canonicalJson"
"relaxedJson"

デフォルトは "relaxedJson" です。

詳しくは、「基本JSONと拡張JSON 」を参照してください。

Basic JSON

メッセージの取り込みを容易にするため、$emit ステージは、処理されたデータを Basic JSON 形式でシンクに書き込むことをサポートします。これにより、Extended（canonicalJson）と Relaxed Extended（relaxedJson）JSON 形式が簡素化されます。Basic JSON では MongoDB Extended JSON ラッパーが使用されないため、すべての BSON types が保存されません。

Basic JSON形式を指定するには、$emit ステージで config.outputFormat フィールドを "basicJson" に設定します。

以下の表は、影響を受けるすべてのフィールドに対するこれらの簡略化の例を示しています。

フィールド型	Relaxed JSON	basicJson
バイナリ	`{ "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}`
ObjectId	`{ "_id": { "$oid": "671a8ce1497407eff0e17cba" }}`	`{ "_id": "6717fcbba18c8a8f74b6d977" }`
Negative Infinity	`{ "negInf": { "$numberDouble": "-Infinity" }}`	`{ "negInf": "-Infinity" }`
Positive Infinity	`{ "posInf": { "$numberDouble": "Infinity" }}`	`{ "posInf": "Infinity" }`
正規表現	`{ "regex": { "$regularExpression": { "pattern": "ab+c", "options": "i" }}}`	`{ "regex": { "pattern": "ab+c", "options": "i" }}`
UUID	`{ "uuid": { "$binary": { "base64": "Kat+fHk6RkuAmotUmsU7gA==", "subType": "04" }}}`	`{ "uuid": "420b7ade-811a-4698-aa64-c8347c719cf1"}`

動作

ストリームプロセッサごとに 1 つの Atlas 時系列コレクションにのみ書込み (write) ができます。存在しないコレクションを指定した場合、Atlas は指定した時系列フィールドでコレクションを作成します。既存のデータベースを指定する必要があります。

Atlas Stream Processing では、$ ステージを使用してAWSまたはAzure Blogストレージシンクからの 16MB より大きいBSONドキュメントの書き込みがサポートされていません。ストリームプロセッサのソースがMongoDBデプロイでない場合、または $lookup などの以前のパイプラインステージによってドキュメントのサイズが増加する場合は、この制限を最大 125MB まで増やすことができます。

Azure Blob Storage $emit バリアントは、 Azure Data Lake Storage v2 コンテナへの書込みもサポートしています。

GCP Pub/Sub$emit orderingKeyattributesバリアントでは、フィールドとフィールドのキーと値のペアの両方のフィールドパス式を介して値を定義できます。式が失敗した場合、または式がstring に変換されない場合、Atlas Stream Processing は影響を受けるメッセージをデッドレターキュー (DLQ)に送信します。

topicストリームプロセッサがメッセージごとに異なるターゲットに書き込むようにするために、フィールドの値としてフィールドまたは動的式を使用できます。式はstring として評価される必要があります。

例

次の形式のメッセージを生成するトランザクションイベントのストリームがあります。

{
  "customer": "Very Important Industries",
  "customerStatus": "VIP",
  "tenantId": 1,
  "transactionType": "subscription"
}
{
  "customer": "N. E. Buddy",
  "customerStatus": "employee",
  "tenantId": 5,
  "transactionType": "requisition"
}
{
  "customer": "Khan Traktor",
  "customerStatus": "contractor",
  "tenantId": 11,
  "transactionType": "billableHours"
}

これらをそれぞれを個別のApache Kafkaトピックに並べ替えるには、次の$emit ステージを記述します。

{
  "$emit": {
    "connectionName": "kafka1",
    "topic": "$customerStatus"
  }
}

この$emitステージ:

Very Important IndustriesメッセージをVIPという名前のトピックに書き込みます。
N. E. Buddyメッセージをemployeeという名前のトピックに書き込みます。
Khan Traktorメッセージをcontractorという名前のトピックに書き込みます。

動的式の詳細については、式演算子を参照してください。

まだ存在しないトピックを指定した場合、 Apache Kafka は、それを対象とする最初のメッセージを受信したときにトピックを自動的に作成します。

動的な式でトピックを指定したものの、Atlas Stream Processing が特定のメッセージの式を評価できない場合、構成されている場合、Atlas Stream Processing はそのメッセージをデッドレターキュー（DLQ）に送信し、以降のメッセージを処理します。デッドレターキュー（DLQ）が構成されていない場合、Atlas Stream Processing はそのメッセージを完全にスキップし、以降のメッセージを処理します。

例

ストリーミングデータソースは、気象用サンプルデータセットのスキーマに準拠して、さまざまな場所から詳細な気象レポートを生成します。次の集計には 3 つのステージがあります。

$sourceステージでは、という名前のトピックでこれらのレポートを収集するApachemy_weatherdata Kafkaプロバイダーとの接続を確立し、各レコードが後続の集計ステージに取り込まれる際に公開します。このステージではまた、プロジェクションを実行するタイムスタンプフィールドの名前が上書きされ、ingestionTime に設定されます。
$matchステージでは、airTemperature.value が以上であるドキュメントを除外し、30.0 がairTemperature.value 30.0未満のドキュメントを次のステージに渡します。
$addFieldsステージは、ストリームをメタデータで強化します。
$emit ステージは、weatherStreamOutput Kafkaブローカー接続を介してstream というトピックに出力を書き込みます。

{
  "$source": {
    "connectionName": "sample_weatherdata",
    "topic": "my_weatherdata",
    "tsFieldName": "ingestionTime"
  }
},
{
  "$match": {
    "airTemperature.value": {
      "$lt": 30
    }
  }
},
{
  "$addFields": {
    "processorMetadata": {
      "$meta": "stream"
    }
  }
},
{
  "$emit": {
    "connectionName": "weatherStreamOutput",
    "topic": "stream"
  }
}

stream トピックのドキュメントは以下の形式をとります：

{
  "st": "x+34700+119500",
  "position": {
    "type": "Point",
    "coordinates": [122.8, 116.1]
  },
  "elevation": 9999,
  "callLetters": "6ZCM",
  "qualityControlProcess": "V020",
  "dataSource": "4",
  "type": "SAO",
  "airTemperature": {
    "value": 6.7,
    "quality": "9"
  },
  "dewPoint": {
    "value": 14.1,
    "quality": "1"
  },
  "pressure": {
    "value": 1022.2,
    "quality": "1"
  },
  "wind": {
    "direction": {
      "angle": 200,
      "quality": "9"
    },
    "type": "C",
    "speed": {
      "rate": 35,
      "quality": "1"
    }
  },
  "visibility": {
    "distance": {
      "value": 700,
      "quality": "1"
    },
    "variability": {
      "value": "N",
      "quality": "1"
    }
  },
  "skyCondition": {
    "ceilingHeight": {
      "value": 1800,
      "quality": "9",
      "determination": "9"
    },
    "cavok": "N"
  },
  "sections": ["AA1", "AG1", "UG1", "SA1", "MW1"],
  "precipitationEstimatedObservation": {
    "discrepancy": "0",
    "estimatedWaterDepth": 999
  },
  "atmosphericPressureChange": {
    "tendency": {
      "code": "4",
      "quality": "1"
    },
    "quantity3Hours": {
      "value": 3.8,
      "quality": "1"
    },
    "quantity24Hours": {
      "value": 99.9,
      "quality": "9"
    }
  },
  "seaSurfaceTemperature": {
    "value": 9.7,
    "quality": "9"
  },
  "waveMeasurement": {
    "method": "M",
    "waves": {
      "period": 8,
      "height": 3,
      "quality": "9"
    },
    "seaState": {
      "code": "00",
      "quality": "9"
    }
  },
  "pastWeatherObservationManual": {
    "atmosphericCondition": {
      "value": "6",
      "quality": "1"
    },
    "period": {
      "value": 3,
      "quality": "1"
    }
  },
  "skyConditionObservation": {
    "totalCoverage": {
      "value": "02",
      "opaque": "99",
      "quality": "9"
    },
    "lowestCloudCoverage": {
      "value": "00",
      "quality": "9"
    },
    "lowCloudGenus": {
      "value": "00",
      "quality": "1"
    },
    "lowestCloudBaseHeight": {
      "value": 1750,
      "quality": "1"
    },
    "midCloudGenus": {
      "value": "99",
      "quality": "1"
    },
    "highCloudGenus": {
      "value": "00",
      "quality": "1"
    }
  },
  "presentWeatherObservationManual": {
    "condition": "52",
    "quality": "1"
  },
  "atmosphericPressureObservation": {
    "altimeterSetting": {
      "value": 1015.9,
      "quality": "9"
    },
    "stationPressure": {
      "value": 1026,
      "quality": "1"
    }
  },
  "skyCoverLayer": {
    "coverage": {
      "value": "08",
      "quality": "1"
    },
    "baseHeight": {
      "value": 2700,
      "quality": "9"
    },
    "cloudType": {
      "value": "99",
      "quality": "9"
    }
  },
  "liquidPrecipitation": {
    "period": 12,
    "depth": 20,
    "condition": "9",
    "quality": "9"
  },
  "extremeAirTemperature": {
    "period": 99.9,
    "code": "N",
    "value": -30.4,
    "quantity": "1"
  },
  "ingestionTime": {
    "$date": "2024-09-26T17:34:41.843Z"
  }
}

注意

前述の例はその一般的な例です。ストリーミングデータは静的ではなく、各ユーザーに異なるドキュメントが表示されます。

戻る

$setStreamMeta

$merge

$emit 集計ステージ（ストリーム プロセシング）

定義

配置

構文

Apache Kafka ブロック

Atlas 時系列コレクション

注意

重要

AWS Kinesis

AWS S3

Azure BLOB ストレージ

Google Cloud パブリック/サブスクライブ

Basic JSON

動作

例

例

注意

`$emit` 集計ステージ（ストリームプロセシング）