定義
$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ステージは、次のフィールドを持つドキュメントを取得します。
フィールド | タイプ | 必要性 | 説明 |
|---|---|---|---|
| string | 必須 | データを取り込む接続の名前（接続レジストリに表示されます）。 |
| string または式 | 必須 | メッセージの送信先Apache Kafkaトピックの名前。 |
| ドキュメント | 任意 | 平均直列化されたソースへの書き込みをサポートするためにスキーマ レジストリの使用を有効にするドキュメント。 この機能を有効にするには、スキーマ Registry 接続を作成する必要があります。 |
| string | 条件付き | Avro 非直列化に使用するスキーマ レジストリ接続の名前。 |
| ドキュメント|式 | 条件付き | 直列化スキーマのプロパティを定義するドキュメント、またはそれを評価する式。 |
| string | 条件付き | スキーマ レジストリを使用する直列化のタイプ。Atlas Stream Processing は現在、スキーマ レジストリ接続を介した |
| ドキュメント | 条件付き | |
| ドキュメント | 任意 | スキーマレジストリ接続のオプションの構成パラメーターを定義するドキュメント。 |
| ブール値 | 任意 | 認識されないスキーマを持つドキュメントをプロセシングする際に、スキーマを自動的に登録するかどうかを判定するトグル。false に設定されている場合、認識されないスキーマを持つドキュメントはデッドレターキュー (DLQ)に送信されます。 デフォルトは |
| string | 条件付き | 自動登録されるスキーマのサブジェクト名を決定する方法。次のいずれかである必要があります。
デフォルトは |
| ドキュメント | 任意 | のさまざまなデフォルト値を上書きするフィールドを含むドキュメント。 |
| string | 任意 | 成功した デフォルト値は
|
| string | 任意 | 制作者が生成するすべてのデータの圧縮形式。デフォルトは「なし」になっています（つまり、圧縮は行われません）。有効な値は次のとおりです。
圧縮はデータの完全なバッチに使用されるため、バッチ処理の効率は圧縮率に影響します。バッチ処理が多いほど、圧縮率が向上します。 |
| string | 任意 | 日付値のための日付フォーマット。有効な値は次のとおりです。
以下に例を挙げます。 次の入力をご覧ください。
|
| 式 | 任意 | 出力メッセージに追加するヘッダー。 実行時に、この式は完全なヘッダーオブジェクト、または 式がオブジェクトと評価される場合、Atlas Stream Processing は、そのオブジェクト内の各キーと値のペアからヘッダーを構築します。キーはヘッダー名、値はヘッダー値です。 式が配列と評価される場合は、キーと値のペア オブジェクトの配列の形式になる必要があります。 例: Atlas Stream Processing は、配列内の各オブジェクトからヘッダーを構築します。キーはヘッダー名で、値はヘッダー値です。Atlas Stream Processing は次の種類のヘッダー値をサポートしています。
フィールド値またはストリームメタデータからヘッダーをビルドする方法を学ぶには、ダイナミック Kafka ヘッダー。を参照してください。 |
| オブジェクト | string | 式 | 任意 | Apache Kafkaメッセージ キーとして評価される 式 。
|
| string | 条件付き | Apache Kafkaキー データを直列化するために使用されるデータ型。次のいずれかの値である必要があります。
デフォルトは |
| string | 任意 | Apache Kafkaにメッセージを発行するときに使用するJSON形式。次のいずれかの値である必要があります。
デフォルトは 詳しくは、「Basic JSON」を参照してください。 |
| 式 | 任意 | Kafkaに 式がブール値値に評価できない場合、または評価できない場合、Atlas Stream Processing はドキュメントをDLQ に書込みます。 この設定を使用すると、 |
Atlas 時系列コレクション
処理されたデータを Atlas 時系列コレクションに書き込むには、次のプロトタイプ形式で$emitパイプライン ステージを使用します。
{ "$emit": { "connectionName": "<registered-connection>", "db": "<target-db>" | <expression>, "coll": "<target-coll>" | <expression>, "parallelism": <number>, "timeseries": { <options> } }}$emitステージは、次のフィールドを持つドキュメントを取得します。
フィールド | タイプ | 必要性 | 説明 |
|---|---|---|---|
| string | 必須 | データを取り込む接続の名前（接続レジストリに表示されます）。 |
| string |式 | 必須 | ターゲット時系列コレクションを含む Atlas データベースに解決される、または式の名前。 |
| string |式 | 必須 | 書込み先の Atlas時系列コレクションに解決される、または式の名前。 |
| integer | 任意 | 時系列シンク演算子が時系列コレクションにデータを書込むために使用する内部書込みスレッドの数。これにより、書込み操作が複数のスレッドに分散され、大容量のワークロードのスループットが向上します。 デフォルトは 設定できる最大値は、ストリームプロセッサーの階層によって異なります。詳しくは、Atlas Stream Processing 階層選択ガイドを参照してください。 各ストリーム プロセッサには、その階層によって決定される最大累積並列処理値があります。ストリーム プロセッサの累積並列処理は、次のように計算されます。
例、 ストリーム プロセッサがその階層の最大累積並列処理を超える場合、Atlas Stream Processing はエラーをスローし、目的のレベルの並列処理に必要な最小プロセッサ階層について提案します。エラーを解決するには、プロセッサをより高い階層に増やすアップするか、ステージの並列処理値を低くする必要があります。詳しくは、Stream Processingをご覧ください。 |
| ドキュメント | 任意 | コレクションの時系列フィールドを定義するドキュメント。 |
注意
時系列コレクション内のドキュメントの最大サイズは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 接続の追加 を参照してください。
処理済みデータを Amazon Web Services 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ステージは、次のフィールドを持つドキュメントを取得します。
フィールド | タイプ | 必要性 | 説明 |
|---|---|---|---|
| string | 必須 | データを取り込む接続の名前（接続レジストリに表示されます）。 |
| string | 必須 | 接続するKinesisデータストリームの名前。 |
| string | 任意 | Kinesis Data Stream が動作するリージョン。AWSは同じ名前の複数のストリームをサポートしており、それぞれが異なるリージョンにあります。このパラメーターにより、Atlas Stream Processing はこのようなストリームを区別することができます。 |
| string |式 | 必須 | 各レコードを受信する Kinesis シャードを決定するために使用されるキー。リテラルの string または string に評価される式である必要があります。 |
| ドキュメント | 任意 | のさまざまなデフォルト値を上書きするフィールドを含むドキュメント。 |
| string | 任意 | Kinesisにメッセージを送信するときに使用するJSON形式。次のいずれかの値である必要があります。
デフォルトは 詳しくは、「Basic JSON」を参照してください。 |
| string | 任意 | 日付値のための日付フォーマット。有効な値は次のとおりです。
以下に例を挙げます。 次の入力をご覧ください。
|
Amazon Web Services S3
処理済みデータを AWS S3 バケットシンク接続に書き込み (write)には、次のプロトタイプ形式で $emit パイプラインステージを使用します：
{ "$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ステージは、次のフィールドを持つドキュメントを取得します。
フィールド | タイプ | 必要性 | 説明 |
|---|---|---|---|
| string | 必須 | 接続レジストリに表示される、データを書き込む先の接続の名前。 |
| string |式 | 必須 | データの書き込み (write)先となるS3バケットの名前。リテラルの string または string を評価する式として指定されます。 |
| string | 任意 | 対象バケットが存在する Amazon Web Services リージョンの名前です。Stream Processing ワークスペースを Amazon Web Services リージョンでホストしている場合、このパラメータはそのリージョンがデフォルトになります。そうでない場合は、Stream Processing ワークスペースのホストリージョンに最も近い AWS リージョンがデフォルトになります。 |
| string |式 | 必須 | S3バケットに書き込まれたオブジェクトのキーのプレフィックス。リテラルのプレフィックス string または、string を評価する式である必要があります。 |
| integer | 任意 | S3 Sink 演算子が S3 にデータを書込むために使用する内部書込みスレッドの数。これにより、書込み操作が複数のスレッドに分散され、大容量のワークロードのスループットが向上します。 デフォルトは 設定できる最大値は、ストリームプロセッサーの階層によって異なります。詳しくは、Atlas Stream Processing 階層選択ガイドを参照してください。 各ストリーム プロセッサには、その階層によって決定される最大累積並列処理値があります。ストリーム プロセッサの累積並列処理は、次のように計算されます。
例、 ストリーム プロセッサがその階層の最大累積並列処理を超える場合、Atlas Stream Processing はエラーをスローし、目的のレベルの並列処理に必要な最小プロセッサ階層について提案します。エラーを解決するには、プロセッサをより高い階層に増やすアップするか、ステージの並列処理値を低くする必要があります。詳しくは、Stream Processingをご覧ください。 |
| 式 | 任意 |
|
| ドキュメント | 任意 | さまざまなデフォルト値を上書きする追加のパラメーターを含むドキュメント。 |
| ドキュメント | 任意 | 書き込み動作を制御する追加パラメーターを含むドキュメントです。これらのパラメーターは、どの閾値が最初に満たされるかに応じて書き込み動作をトリガーします。 例えば、取り込まれたドキュメントが |
| integer | 任意 | S3 に書き込む各ファイルにグループ化するドキュメントの数。 |
| integer | 任意 | S3 にファイルを書き込む前に蓄積される必要がある最小バイト数を指定します。バイト数は、最終出力ファイルのサイズではなく、パイプラインで取り込まれた BSON ドキュメントのサイズによって決まります。 デフォルトは 32 MB です。 |
| ドキュメント | 任意 | ドキュメントを一括書き込みするためのタイマーを デフォルトは 1 分です。いかなる |
| integer | 条件付き | ストリームプロセッサが S3 にドキュメントを一括書き込みした後の デフォルトは |
| string | 条件付き | 一括書き込みタイマーをカウントする時間の単位です。このパラメーターは以下の値をサポートしています：
デフォルトは |
| string | 任意 | 出力されたファイル内の各エントリの間の区切り文字。 デフォルトは |
| string | 任意 | |
| string | 任意 | 日付値のための日付フォーマット。有効な値は次のとおりです。
例として、次のレコードをパイプラインに追加した場合:
|
| string | 任意 | 使用する圧縮アルゴリズムの名称。次のいずれかの値でなければなりません：
|
| integer | 条件付き | 出力されるメッセージに適用する圧縮レベルです。 デフォルトは このパラメータは、 |
Azure BLOB ストレージ
処理済みデータを Azure Blob ストレージに書き込み (write)するには、次のプロトタイプ形式を使用して $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ステージは、次のフィールドを持つドキュメントを取得します。
フィールド | タイプ | 必要性 | 説明 |
|---|---|---|---|
| string | 必須 | データを書き込む接続の名前（接続レジストリに表示）。 |
| string | 必須 | ブロブを書き込むストレージコンテナ。 |
| string |式 | 必須 | Azure Blob Storage に書き込まれたオブジェクトのキーのプレフィックス。リテラル string または string を評価する式である必要があります。 |
| ドキュメント | 任意 | さまざまなデフォルト値を上書きする追加のパラメーターを含むドキュメント。 |
| string | 任意 | 日付値のための日付フォーマット。有効な値は次のとおりです。
例として、次のレコードをパイプラインに追加した場合:
|
| string | 任意 | 使用する圧縮アルゴリズムの名称。次のいずれかの値でなければなりません：
このフィールドに値を指定しない場合、Atlas Stream Processing は圧縮なしでデータを書込みます。 |
| integer | 条件付き | 出力されるメッセージに適用する圧縮レベルです。 デフォルトは このパラメータは、 |
| string | 任意 | |
| string | 任意 | 出力されたファイル内の各エントリの間の区切り文字。 デフォルトは |
| ドキュメント | 任意 | 書き込み動作を制御する追加パラメーターを含むドキュメントです。これらのパラメーターは、どの閾値が最初に満たされるかに応じて書き込み動作をトリガーします。 例えば、取り込まれたドキュメントが |
| integer | 任意 | Azure Blob Storage に書き込まれた各ファイルにグループ化するドキュメントの数。 |
| integer | 任意 | ファイルがAzure Blob ストレージに書き込まれる前に蓄積する必要がある最小バイト数を指定します。バイト数は、最終出力ファイルのサイズではなく、パイプラインによって取り込まれたBSONドキュメント のサイズによって決まります。 デフォルトは 32 MB です。 |
| ドキュメント | 任意 | ドキュメントを一括書き込みするためのタイマーを デフォルトは 1 分です。いかなる |
| integer | 条件付き |
デフォルトは |
| string | 条件付き | 一括書き込みタイマーをカウントする時間の単位です。このパラメーターは以下の値をサポートしています：
デフォルトは |
| integer | 任意 | Atlas Stream Processing が Azure Blob ストレージにデータを書き込むために使用する内部書込みスレッドの数。これにより、書込み操作が複数のスレッドに分散され、大容量のワークロードのスループットが向上します。 デフォルトは |
| string |式 | 任意 |
|
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ステージは、次のフィールドを持つドキュメントを取得します。
フィールド | タイプ | 必要性 | 説明 |
|---|---|---|---|
| string | 必須 | データを書き込む接続の名前（接続レジストリに表示）。 |
| string | 必須 | 公開するプロジェクトのID 。 |
| string |式 | 必須 | メッセージを公開する特定のプロジェクト内のトピックの名前。string、または string を評価する式である必要があります。 |
| string | 必須 | リクエストの送信先となるロケーション エンドポイント。利用可能なリージョンについては、 Pub/Sub のドキュメントを参照してください。 |
| string |式 | 任意 | メッセージの公開を順序付ける公開データ内のキー。If you set an このフィールドに string 値を使用して、静的 |
| ドキュメント | 任意 | メッセージに追加する属性を定義するドキュメント。属性はキーと値のペアの形式をとります。ここでは、
|
| ドキュメント | 任意 | さまざまなデフォルト値を上書きする追加のパラメーターを含むドキュメント。 |
| string | 任意 | 日付値のための日付フォーマット。有効な値は次のとおりです。
例として、次のレコードをパイプラインに追加した場合:
|
| string | 任意 |
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 |
|---|---|---|
バイナリ |
|
|
日付 |
|
|
小数点 |
|
|
タイムスタンプ |
|
|
ObjectId |
|
|
Negative Infinity |
|
|
Positive Infinity |
|
|
正規表現 |
|
|
UUID |
|
|
動作
ストリーム プロセッサごとに 1 つの Atlas 時系列コレクションにのみ書込み (write) ができます。 存在しないコレクションを指定した場合、Atlas は指定した時系列フィールドでコレクションを作成します。 既存のデータベースを指定する必要があります。
Atlas Stream Processingは、$emit ステージを使用して、16MB を超えるBSON ドキュメントをAmazon Web Services または Azure ブロブ ストレージ シンクに書き込むことをサポートしていません。ストリームプロセッサーのソースがMongoDB配置でない場合、または、$lookupなどの前のパイプラインステージでドキュメントのサイズが増加する場合は、この制限を125MBまで増やすことができます。
この Azure Blob ストレージ $emit バリアントは、Azure Datalake ストレージ v2 コンテナへの書き込みもサポートしています。
GCP Pub/Sub $emit バリアントでは、orderingKeyフィールドと attributesフィールドのキーと値のペアの両方のフィールドパスパス式を介して値を定義できます。式が失敗した場合、または式がstring に変換されない場合、Atlas Stream Processing は影響を受けるメッセージをデッドレターキュー (DLQ)に送信します。
ストリーム プロセッサがメッセージごとに異なるターゲットへ書き込めるようにするには、topic フィールドの値としてフィールドまたは動的式を使用します。この式は文字列として評価される必要があります。
例
次の形式のメッセージを生成するトランザクション イベントのストリームがあります。
{ "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 はそのメッセージを完全にスキップし、以降のメッセージを処理します。
動的 Kafka ヘッダー
Apache Kafkaヘッダーを動的に構築するには、前のパイプラインステージで完全なヘッダーオブジェクトをビルドするし、$emit.config.headersからそのオブジェクトを参照します。ヘッダーが送信メタデータであり、出力メッセージの本文に含めるべきではない場合は、$setStreamMetaを使用します。
{ "$setStreamMeta": { "stream.kafka_headers": { "$arrayToObject": [ { "k": "test_header", "v": "$myField" } ] } }},{ "$emit": { "connectionName": "kafka", "topic": "my-topic", "config": { "headers": { "$meta": "stream.kafka_headers" } } }}例
このセクションの例を実行するには、Stream Processingワークスペースを作成し、Atlas Stream Processingワークスペースへの接続を追加します。学ぶ方法については、「Atlas Stream Processingの開始方法」および「Atlas Stream Processing接続の追加」を参照してください。
Kafka トピックへの書き込み (write)
ストリーミング データソースは、気象用サンプル データセットのスキーマに準拠して、さまざまなロケーションから詳細な気象レポートを生成します。次の集計には 3 つのステージがあります。
$sourceステージでは、 という名前のトピックでこれらのレポートを収集するApachemy_weatherdataKafkaプロバイダーとの接続を確立し、各レコードが後続の集計ステージに取り込まれる際に公開します。このステージではまた、プロジェクションを実行するタイムスタンプフィールドの名前が上書きされ、ingestionTimeに設定されます。$matchステージでは、airTemperature.valueが30.0以上であるドキュメントを除外し、airTemperature.valueが30.0未満のドキュメントを次のステージに渡します。$addFieldsステージでは、ストリームにメタデータが追加されます。$emitステージは、weatherStreamOutputKafkaブローカー接続を介して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" }}注意
前述の例はその一般的な例です。 ストリーミング データは静的ではなく、各ユーザーに異なるドキュメントが表示されます。
時系列コレクションへの書き込み (write)
次の集計では、サンプルストリーミングデータソース sample_stream_solar からレポートを読み取り、時系列コレクションに書き込み (write) ます。この集計には 3 つのステージがあります。
$sourceステージは、sample_stream_solarソースに接続します。$setステージは、各レポートのtimestampフィールドを$dateFromStringを使用して string から日付に変換します。これにより、時系列コレクションでそれをtimeFieldとして使用できるようになります。$emitステージは、各レポートをsolar-cluster-connection接続経由でsolarDbデータベースのsolar_ts時系列コレクションに書き込み (write) ます。metaFieldオプションはdevice_idに設定されているため、同じデバイスからのレポートはバケットにまとめて保存 (する) されます。solar_tsコレクションが存在しない場合、Atlas Stream Processing はそれを時系列コレクションとして作成します。
{ "$source": { "connectionName": "sample_stream_solar" }},{ "$set": { "timestamp": { "$dateFromString": { "dateString": "$timestamp" } } }},{ "$emit": { "connectionName": "solar-cluster-connection", "db": "solarDb", "coll": "solar_ts", "timeseries": { "timeField": "timestamp", "metaField": "device_id" } }}注意
sample_stream_solar ソースは、迅速なプロトタイピングのために毎秒サンプルドキュメントを生成するテスト専用の接続です。
solar_ts コレクションのドキュメントは以下の形式をとります：
{ "timestamp": { "$date": "2026-10-22T15:07:49.299Z" }, "device_id": "device_0", "max_watts": 450, "_id": { "$oid": "6ada26c5ad54068b03089593" }, "obs": { "watts": 135, "temp": 5 }, "event_type": 0, "group_id": 8}注意
前述の例はその一般的な例です。 ストリーミング データは静的ではなく、各ユーザーに異なるドキュメントが表示されます。