$merge (Processamento de Stream)

Limitações

The on field has special requirements for $merge against sharded collections. To learn more, see $merge syntax.

$merge can't write to time series collections. To write documents to time series collections, use the $emit stage.

Você precisa ter a função de administrador do Atlas para usar $merge em coleções fragmentadas.

Expressões Dinâmicas

You can use a field or dynamic expression as the value of the following fields:

• into.db

• into.coll

Isso permite que seu processador de fluxo grave mensagens em diferentes collection de destino do Atlas, mensagem por mensagem.

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

$merge: {
  into: {
    connectionName: "db1",
    db: "$customerStatus",
    coll: "$transactionType"
  }
}

You can only use dynamic expressions that evaluate to strings. For more information on dynamic expressions, see expression operators.

If you specify a database or collection with a dynamic expression, but Atlas Stream Processing cannot evaluate the expression for a given message, Atlas Stream Processing sends that message to the dead letter queue if configured and processes subsequent messages. If there is no dead letter queue configured, then Atlas Stream Processing skips the message completely and processes subsequent messages.

Salvando dados de tópicos do Kafka

Para salvar dados de streaming de múltiplos tópicos do Apache Kafka em coleções no seu cluster Atlas, utilize o estágio $merge com o estágio $source. A etapa $source especifica os tópicos dos quais ler dados. O estágio $merge grava os dados na coleção de destino.

Use a seguinte sintaxe:

{
  "$source": {
    "connectionName": "<registered-kafka-connection>",
    "topic": [ "<topic-name-1>", "<topic-name-2>", ... ]
  }
},
{
  "$merge": {
    "into": {
      "connectionName": "<registered-atlas-connection>",
      "db": "<registered-database-name>" | <expression>,
      "coll": "<atlas-collection-name>" | <expression>
      }
    },
    ...
}

Exemplo básico

A streaming data source generates detailed weather reports from various locations, conformant to the schema of the Sample Weather Dataset. The following aggregation has three stages:

1. O estágio $source estabelece uma conexão com o broker do Apache Kafka coletando esses relatórios em um tópico chamado my_weatherdata, expondo cada registro à medida que ele é assimilado nos estágios de agregação subsequentes. Esse estágio também substitui o nome do campo de carimbo de data/hora que ele projeta, definindo-o como ingestionTime.

2. The $match stage excludes documents that have a dewPoint.value of less than or equal to 5.0 and passes along the documents with dewPoint.value greater than 5.0 to the next stage.

3. The $merge stage writes the output to an Atlas collection named stream in the sample_weatherstream database. If no such database or collection exist, Atlas creates them.

{
  '$source': {
    connectionName: 'sample_weatherdata',
    topic: 'my_weatherdata',
    tsFieldName: 'ingestionTime'
  }
},
{ '$match': { 'dewPoint.value': { '$gt': 5 } } },
{
  '$merge': {
    into: {
      connectionName: 'weatherStreamOutput',
      db: 'sample_weatherstream',
      coll: 'stream'
    }
  }
}

Para visualizar os documentos na coleção sample_weatherstream.stream resultante, conecte-se ao cluster Atlas e execute o seguinte comando:

db.getSiblingDB("sample_weatherstream").stream.find()

{
  _id: ObjectId('66ad2edfd4fcac13b1a28ce3'),
  airTemperature: { quality: '1', value: 27.7 },
  atmosphericPressureChange: {
    quantity24Hours: { quality: '9', value: 99.9 },
    quantity3Hours: { quality: '1' },
    tendency: { code: '1', quality: '1' }
  },
  atmosphericPressureObservation: {
    altimeterSetting: { quality: '1', value: 1015.9 },
    stationPressure: { quality: '1', value: 1021.9 }
  },
  callLetters: 'CGDS',
  dataSource: '4',
  dewPoint: { quality: '9', value: 25.7 },
  elevation: 9999,
  extremeAirTemperature: { code: 'N', period: 99.9, quantity: '9', value: -30.4 },
  ingestionTime: ISODate('2024-08-02T19:09:18.071Z'),
  liquidPrecipitation: { condition: '9', depth: 160, period: 24, quality: '2' },
  pastWeatherObservationManual: {
    atmosphericCondition: { quality: '1', value: '8' },
    period: { quality: '9', value: 3 }
  },
  position: { coordinates: [ 153.3, 50.7 ], type: 'Point' },
  precipitationEstimatedObservation: { discrepancy: '4', estimatedWaterDepth: 4 },
  presentWeatherObservationManual: { condition: '53', quality: '1' },
  pressure: { quality: '1', value: 1016.3 },
  qualityControlProcess: 'V020',
  seaSurfaceTemperature: { quality: '9', value: 27.6 },
  sections: [ 'AA2', 'SA1', 'MW1', 'AG1', 'GF1' ],
  skyCondition: {
    cavok: 'N',
    ceilingHeight: { determination: 'C', quality: '1', value: 6900 }
  },
  skyConditionObservation: {
    highCloudGenus: { quality: '1', value: '05' },
    lowCloudGenus: { quality: '9', value: '03' },
    lowestCloudBaseHeight: { quality: '9', value: 150 },
    lowestCloudCoverage: { quality: '1', value: '05' },
    midCloudGenus: { quality: '9', value: '08' },
    totalCoverage: { opaque: '99', quality: '1', value: '06' }
  },
  skyCoverLayer: {
    baseHeight: { quality: '9', value: 99999 },
    cloudType: { quality: '9', value: '05' },
    coverage: { quality: '1', value: '04' }
  },
  st: 'x+35700-027900',
  type: 'SAO',
  visibility: {
    distance: { quality: '1', value: 4000 },
    variability: { quality: '1', value: 'N' }
  },
  waveMeasurement: {
    method: 'I',
    seaState: { code: '99', quality: '9' },
    waves: { height: 99.9, period: 14, quality: '9' }
  },
  wind: {
    direction: { angle: 280, quality: '9' },
    speed: { quality: '1', rate: 30.3 },
    type: '9'
  }
}

Replicar eventos do change stream nos Atlas clusters

You can use the $merge.whenMatched and $merge.whenNotMatched parameters to replicate the effects of Change Stream events according to their operation type.

A seguinte agregação tem quatro fases:

1. O estágio $source estabelece uma conexão com a coleção db1.coll1 em um cluster do Atlas por meio da conexão atlas1.

2. The $addFields stage enriches the ingested documents with a fullDocument._isDelete field set to the value of an equality check between the "$operationType value of each document and "delete". This equality evaluates to a boolean.

3. O estágio $replaceRoot substitui o documento pelo valor do campo $fullDocument enriquecido.

4. O estágio $merge grava em db1.coll1 através da conexão atlas2, realizando duas verificações em cada documento:

   • Primeiro, o campo whenMatched verifica se o documento corresponde a um documento existente na coleção db1.coll1 usando _id, que é o campo de correspondência padrão, já que on não está explicitamente definido. Se isso acontecer e fullDocument._isDelete estiver definido como true, então o Atlas exclui o documento correspondente. Se corresponder e fullDocument._isDelete estiver definido como false, o Atlas substituirá o documento correspondente pelo novo da fonte de dados de streaming.

   • Em segundo lugar, se o Atlas Stream Processing não encontrar tal documento correspondente e fullDocument._isDelete for verdadeiro, o Atlas descartará o documento em vez de gravá-lo na coleção. Se não houver um documento correspondente e fullDocument._isDelete for falso, o Atlas insere o documento da fonte de dados de streaming na coleção.

{
  $source: {
    connectionName: “atlas1”,
    db: “db1”,
    coll: “coll1”,
    fullDocument: “required”
  }
},
{
  $addFields: {
    “fullDocument._isDelete”: {
      $eq: [
        “$operationType”,
        “delete”
      ]
    }
  }
},
{
  $replaceRoot: {
    newRoot: “$fullDocument”
  }
},
{
  $merge: {
    into: {
      connectionName: “atlas2”,
      db: “db1”,
      coll: “coll1”
    },
    whenMatched: {
      $cond: {
        if: “$_isDelete”,
        then: “delete”,
        else: “replace”
      }
    },
    whenNotMatched: {
      $cond: {
        if: “$_isDelete”,
        then: “discard”,
        else: “insert”
      }
    },
  }
}