`$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 つのステージがあります。

ステージは$source Apache Kafka との接続を確立しますエージェントは、my_weatherdata という名前のトピックでこれらのレポートを収集し、各レコードが取り込まれるときに後続の集計ステージに渡します。
$validateステージでは、ドキュメントにフィールドとフィールドの配列値があるかどうかがチェックされ、配列値を持つドキュメントはパイプラインの残りの部分に渡され、ないドキュメントはposition.coordinates sectionsDLQ に渡されます。
$matchステージでは、 wind.speed.rateの値が30以上であるドキュメントを除外し、 wind.speed.rateの値が30未満のドキュメントを渡します。
$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.coordinatesとsectionsに期待される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.coordinates 、 sections 、またはその両方の無効な値があることを確認します。

戻る

$source

$ https