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

string

必須

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

string |式

必須

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

ドキュメント

任意

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

整数

任意

成功した $emit 操作に対して、Apache Kafka クラスターから求められる確認応答の数。

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

• -1

• 0

• 1

• all

string

任意

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

• none

• gzip

• snappy

• lz4

• zstd

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

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

式

任意

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

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

オブジェクト | string | 式

任意

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

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

string

条件付き

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

• "binData"

• "string"

• "json"

• "int"

• "long"

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

string

任意

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

• "relaxedJson"

• "canonicalJson"

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

式

任意

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トピックの圧縮はトリガーされません。

string

必須

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

string

必須

データを書き込む先の S3 バケットの名前。

string

任意

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

string |式

必須

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

ドキュメント

任意

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

ドキュメント

任意

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

例えば、取り込まれたドキュメントが config.writeOptions.interval の閾値に達していなくても config.writeOptions.count の閾値に達した場合、ストリームプロセッサは config.writeOptions.count の閾値に従ってこれらのドキュメントを S3 に送出します。

integer

任意

S3 に書き込む各ファイルにグループ化するドキュメントの数。

integer

任意

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

ドキュメント

任意

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

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

integer

条件付き

ストリームプロセッサが S3 にドキュメントを一括書き込みするまでの、writeOptions.interval.units で指定された単位数。

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

string

条件付き

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

• ms

• second

• minute

• hour

• day

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

string

任意

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

デフォルトは \n です。

string

任意

S3 に書き込まれる JSON の出力形式を指定します。次のいずれかの値でなければなりません：

• "basicJson"

• "canonicalJson"

• "relaxedJson"

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

詳しくは、Basic JSON をご覧ください。

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

string

任意

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

• "gzip"

• "snappy"

string

条件付き

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

デフォルトは 6 です。

このパラメーターは gzip に対して必須であり、かつそれに限定されます。config.compression を snappy に設定した場合、このパラメーターを設定しても効果はありません。

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

{ "_id": { "$oid": "671a8ce1497407eff0e17cba" }}

{ "_id": "6717fcbba18c8a8f74b6d977" }

{ "negInf": { "$numberDouble": "-Infinity" }}

{ "negInf": "-Infinity" }

{ "posInf": { "$numberDouble": "Infinity" }}

{ "posInf": "Infinity" }

{ "regex": { "$regularExpression": { "pattern": "ab+c", "options": "i" }}}

{ "regex": { "pattern": "ab+c", "options": "i" }}

{ "uuid": { "$binary": { "base64": "Kat+fHk6RkuAmotUmsU7gA==", "subType": "04" }}}

{ "uuid": "420b7ade-811a-4698-aa64-c8347c719cf1"}