Docs Menu
Docs Home
/
MongoDB Atlas
/ /

$validate

項目一覧

  • 定義
  • 構文
  • 動作
  • バリデーターの例

$validateステージでは、ストリーミング ドキュメントが予想される範囲、値、またはデータ型のスキーマに準拠しているかどうかがチェックされます。

$validate

$validateパイプライン ステージには次のプロトタイプ形式があります。

{
"$validate": {
"validator": { <filter> },
"validationAction" : "discard" | "dlq"
}
}

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

フィールド
タイプ
必要性
説明
validator
ドキュメント
必須

ユーザー定義のスキーマに対して受信メッセージを検証するために使用される式のドキュメント。 検証式を定義するには、次のクエリ演算子を除くすべてのクエリ演算子を使用できます。

  • $near

  • $nearSphere

  • $text

  • $where

validationAction
string
任意

メッセージがユーザー定義のスキーマに違反した場合に実行するアクションを指定します。 次のいずれかの値を指定できます。

  • discard: メッセージを破棄します。 validationActionの値を指定しない場合、これはデフォルトの動作になります。

  • dlq: ストリーム プロセッサ構成で定義されたコレクションに違反をログに記録し、トランザクションの保証なしでベストエフォートの破棄を実行します。

$validate$sourceステージの後、および$emit $mergeステージまたは ステージの前のパイプライン内の任意の点で を使用できます。

validationActionフィールドにdiscardまたはdlqオプションのいずれかを指定すると、Atlas Stream Processing は検証に失敗したメッセージを次の形式でログに記録します。

{
"t": <datetime>,
"s": "<severity-level>",
"c": "streams-<job-name>",
"ctx": "<processed-pipeline>",
"msg": "<message-body>",
"attrs": {
<result-of-logAttributes-evaluation>
},
"tags": <array-of-strings>,
"truncated": {
<truncation-description>
},
"size": <size-of-entry>
}

以下の表は、ログエントリのフィールドを説明したものです。

フィールド
タイプ
説明
attrs
ドキュメント
$validate定義内のlogAttributesフィールドを評価した結果を含むドキュメント。 結果は、フィールドのリストです。
c
string
失敗が発生した特定のストリーム プロセシング ジョブの名前。
ctx
string
処理されているストリーミング データ パイプラインの名前。
msg
string
検証に失敗したメッセージの本文。

Atlas Stream Processing は JSON schema ドラフト4 のみをサポートしています またはそれ以前

次のドキュメントは、 $andを使用して論理 AND 操作を実行するバリデーター式の例を示しています。

{
$validate: {
validator: {
$and: [{
$expr: {
$ne: [
"$Racer_Name",
"Pace Car"
]
}
},
{
$jsonSchema: {
required: [ "Racer_Num", "Racer_Name", "lap", "Corner_Num", "timestamp" ],
properties: {
Racer_Num: {
bsonType: "int",
description: "'Racer_Num' is the integer number of the race car and is required"
},
Racer_Name: {
bsonType: "string",
description: "'Racer_Name' must be a string and is required"
},
lap: {
bsonType: "int",
minimum: 1,
description: "'lap' must be a int and is required"
},
Corner_Num: {
bsonType: "int",
minimum: 1,
maximum: 4,
description: "'Corner_Num' must be a int between 1 and 4 and is required"
},
timestamp: {
bsonType: "string",
pattern: "^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{6}$",
description: "'timestamp' must be a string matching iso date pattern and is required"
}
}
}
}]
}, validationAction : "dlq"
}
}

さまざまな場所から詳細な天気予報を生成するストリーミング データソースを検討します。 次の集計パイプラインの例には、ドキュメントが$validate サンプル 気象 データセット のスキーマに準拠しているようにするために、 ステージを含めています。集計には 4 つのステージがあります。

  1. ステージは$source Apache Kafka との接続を確立します エージェントは、my_weatherdata という名前のトピックでこれらのレポートを収集し、各レコードが取り込まれるときに後続の集計ステージに渡します。

  2. $validateステージでは、ドキュメントに フィールドと フィールドの配列値があるかどうかがチェックされ、配列値を持つドキュメントはパイプラインの残りの部分に渡され、ないドキュメントはposition.coordinates sectionsDLQ に渡されます。

  3. $matchステージでは、 wind.speed.rateの値が30以上であるドキュメントを除外し、 wind.speed.rateの値が30未満のドキュメントを渡します。

  4. $mergeステージは、 sample_weatherstreamデータベース内のstreamという名前の Atlas コレクションに出力を書き込みます。 そのようなデータベースやコレクションが存在しない場合は、Atlas がそれらを作成します。

{
'$source': {
connectionName: 'sample_weatherdata',
topic: 'my_weatherdata',
tsFieldName: 'ingestionTime'
}
},
{
'$validate': {
validator: {
'$jsonSchema': { properties: { position: [Object], sections: [Object] } }
},
validationAction: 'dlq'
}
},
{ '$match': { 'wind.speed.rate': { '$lt': 30 } } },
{
'$merge': {
into: {
connectionName: 'weatherStreamOutput',
db: 'sample_weatherstream',
coll: 'stream'
}
}
}

結果のsample_weatherstream.sampleコレクション内のドキュメントを表示するには、 mongoshを使用して Atlas クラスターに接続し、次のコマンドを実行します。

注意

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

db.getSiblingDB("sample_weatherstream").stream.find()
{
_id: ObjectId('66b25302fe8bbac5f39dbdba'),
_stream_meta: {
source: {
type: 'kafka',
topic: 'my_weatherdata',
partition: 0,
offset: Long('168843')
}
},
airTemperature: { quality: '9', value: 3.5 },
atmosphericPressureChange: {
quantity24Hours: { quality: '9', value: 99.9 },
quantity3Hours: { quality: '1', value: 10.9 },
tendency: { code: '3', quality: '1' }
},
atmosphericPressureObservation: {
altimeterSetting: { quality: '1', value: 1015.9 },
stationPressure: { quality: '9', value: 1022.5 }
},
callLetters: 'JIVX',
dataSource: '4',
dewPoint: { quality: '9', value: 20.5 },
elevation: 9999,
extremeAirTemperature: { code: 'N', period: 99.9, quantity: '9', value: -30.4 },
ingestionTime: ISODate('2024-08-06T16:44:50.322Z'),
liquidPrecipitation: { condition: '9', depth: 7000, period: 12, quality: '9' },
pastWeatherObservationManual: {
atmosphericCondition: { quality: '1', value: '7' },
period: { quality: '1', value: 3 }
},
position: { coordinates: [ 120.7, -98.2 ], type: 'Point' },
precipitationEstimatedObservation: { discrepancy: '5', estimatedWaterDepth: 999 },
presentWeatherObservationManual: { condition: '90', quality: '1' },
pressure: { quality: '1', value: 1028.2 },
qualityControlProcess: 'V020',
seaSurfaceTemperature: { quality: '9', value: 11.1 },
sections: [ 'UG1', 'MA1', 'GA3', 'KA1', 'UA1' ],
skyCondition: {
cavok: 'N',
ceilingHeight: { determination: 'C', quality: '1', value: 390 }
},
skyConditionObservation: {
highCloudGenus: { quality: '1', value: '06' },
lowCloudGenus: { quality: '9', value: '07' },
lowestCloudBaseHeight: { quality: '1', value: 800 },
lowestCloudCoverage: { quality: '9', value: '06' },
midCloudGenus: { quality: '9', value: '07' },
totalCoverage: { opaque: '99', quality: '1', value: '99' }
},
skyCoverLayer: {
baseHeight: { quality: '9', value: 1200 },
cloudType: { quality: '9', value: '04' },
coverage: { quality: '1', value: '09' }
},
st: 'x+36700+144300',
type: 'FM-13',
visibility: {
distance: { quality: '9', value: 9000 },
variability: { quality: '9', value: 'N' }
},
waveMeasurement: {
method: 'I',
seaState: { code: '00', quality: '9' },
waves: { height: 9.5, period: 4, quality: '9' }
},
wind: {
direction: { angle: 140, quality: '1' },
speed: { quality: '2', rate: 15.9 },
type: 'N'
}
}

このコレクション内のすべてのドキュメントに、 position.coordinatessectionsに期待されるarrayタイプの値があることを確認します。 dlqという名前のデッド レター キューがある場合、検証に失敗したドキュメントを表示するには、次のコマンドを実行します。

db.getSiblingDB("sample_weatherstream").dlq.find()
{
_id: ObjectId('66b254d3a045fb1406047394'),
_stream_meta: {
source: {
type: 'kafka',
topic: 'my_weatherdata',
partition: 0,
offset: Long('168949'),
key: Binary.createFromBase64('', 0),
headers: []
}
},
errInfo: { reason: 'Input document found to be invalid in $validate stage' },
doc: {
airTemperature: { quality: '9', value: 7.6 },
atmosphericPressureChange: {
quantity24Hours: { quality: '9', value: 99.9 },
quantity3Hours: { quality: '1', value: 0.3 },
tendency: { code: '8', quality: '1' }
},
atmosphericPressureObservation: {
altimeterSetting: { quality: '9', value: 1015.9 },
stationPressure: { quality: '1', value: 1017 }
},
callLetters: 'WRGL',
dataSource: '4',
dewPoint: { quality: '9', value: 25.3 },
elevation: 9999,
extremeAirTemperature: { code: 'M', period: 99.9, quantity: '1', value: -30.9 },
liquidPrecipitation: { condition: '9', period: 99, quality: '9' },
pastWeatherObservationManual: {
atmosphericCondition: { quality: '1', value: '2' },
period: { quality: '1', value: 6 }
},
position: { coordinates: -100.2, type: 'Point' },
precipitationEstimatedObservation: { discrepancy: '5', estimatedWaterDepth: 17 },
presentWeatherObservationManual: { condition: '08', quality: '1' },
pressure: { quality: '9', value: 1001 },
qualityControlProcess: 'V020',
seaSurfaceTemperature: { quality: '9', value: 10.4 },
sections: [ 'GA2', 'GA1', 'KA1', 'AA1', 'OA1' ],
skyCondition: {
cavok: 'N',
ceilingHeight: { determination: 'C', quality: '1', value: 240 }
},
skyConditionObservation: {
highCloudGenus: { quality: '1', value: '02' },
lowCloudGenus: { quality: '9', value: '02' },
lowestCloudBaseHeight: { quality: '1', value: 150 },
lowestCloudCoverage: { quality: '1', value: '03' },
midCloudGenus: { quality: '1', value: '06' },
totalCoverage: { opaque: '99', quality: '1', value: '06' }
},
skyCoverLayer: {
baseHeight: { quality: '9', value: 450 },
cloudType: { quality: '9', value: '03' },
coverage: { quality: '1', value: '07' }
},
st: 'x+20500-074300',
type: 'SAO',
visibility: {
distance: { quality: '9', value: 3800 },
variability: { quality: '9', value: 'N' }
},
waveMeasurement: {
method: 'I',
seaState: { code: '00', quality: '9' },
waves: { height: 37.5, period: 7, quality: '9' }
},
wind: {
direction: { angle: 230, quality: '1' },
speed: { quality: '1', rate: 46.3 },
type: 'N'
},
ingestionTime: ISODate('2024-08-06T16:52:35.287Z'),
_stream_meta: {
source: {
type: 'kafka',
topic: 'my_weatherdata',
partition: 0,
offset: Long('168949')
}
}
},
processorName: 'sampleWeather'
}

デッド レター キュー内のすべてのドキュメントに、 position.coordinatessections 、またはその両方の無効な値があることを確認します。

戻る

$source