$merge (Stream Processing)

Limitaciones

El campo on tiene requisitos especiales para $merge contra colecciones particionadas. Para obtener más información, consulte la sintaxis de $merge.

$merge no se puede guardar en colecciones de series de tiempo. Para guardar documentos en colecciones de series de tiempo, utiliza la etapa $emit.

Debes tener rol de administrador de Atlas para utilizar $merge en colecciones particionadas.

Expresiones dinámicas

Puedes utilizar una expresión dinámica como valor de los siguientes campos:

• into.db

• into.coll

Esto permite que tu procesador de flujo escriba mensajes en diferentes colecciones de Atlas objetivo en función de cada mensaje.

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

Solo se pueden usar expresiones dinámicas que se evalúen como cadenas de texto. Para obtener más información sobre expresiones dinámicas, consulta los operadores de expresiones.

Si especificas una base de datos o colección con una expresión dinámica, pero Atlas Stream Processing no puede evaluar la expresión para un mensaje dado, Atlas Stream Processing envía ese mensaje a la fila de letra muerta si está configurada y procesa los mensajes siguientes. Si no se configura una fila de letra muerta, entonces Atlas Stream Processing omite el mensaje por completo y procesa los mensajes siguientes.

Guardando datos de temas de Kafka

Para guardar datos de transmisión de varios Temas de Apache Kafka en colecciones de tu clúster de Atlas, utiliza la etapa $merge junto con la etapa $source. La etapa $source especifica los temas de los que se debe leer datos. La etapa $merge escribe los datos en la colección objetivo.

Utiliza la siguiente sintaxis:

{
  "$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>
      }
    },
    ...
}

Ejemplo básico

Una fuente de datos de transmisión genera informes meteorológicos detallados desde varias ubicaciones, cumpliendo con el esquema del Conjunto de Datos Meteorológicos de Muestra. La siguiente agregación tiene tres etapas:

1. La etapa $source establece una conexión con el Apache Kafka broker que recopila estos informes en un tema denominado my_weatherdata, exponiendo cada registro a medida que se procesa a las siguientes etapas de agregación. Este etapa también reemplaza el nombre del campo de marca de tiempo que proyecta, estableciéndolo en ingestionTime.

2. La etapa $match excluye los documentos que tienen un dewPoint.value menor o igual a 5.0 y pasa los documentos con dewPoint.value mayor que 5.0 a la siguiente etapa.

3. La fase $merge escribe la salida en una colección Atlas denominada stream en la base de datos sample_weatherstream. Si no existe tal base de datos o colección, Atlas los creará.

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

Para ver los documentos de la colección sample_weatherstream.stream resultante, conéctate a un clúster de Atlas y ejecuta el siguiente 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 del flujo de cambios en clústeres de Atlas

Puede utilizar los parámetros $merge.whenMatched y $merge.whenNotMatched para replicar los efectos de los eventos de Cambio de stream según su tipo de operación.

La siguiente agregación tiene cuatro etapas:

1. La etapa $source establece una conexión a la colección db1.coll1 en un clúster de Atlas a través de la conexión atlas1.

2. La etapa $addFields enriquece los documentos ingresados con un campo fullDocument._isDelete establecido en el valor de una verificación de igualdad entre el valor "$operationType de cada documento y "delete". Esta igualdad evalúa a un booleano.

3. La etapa $replaceRoot sustituye el documento con el valor del campo enriquecido $fullDocument.

4. La etapa $merge escribe en db1.coll1 a través de la conexión atlas2, realizando dos comprobaciones en cada documento:

   • Primero, el campo whenMatched verifica si el documento coincide con un documento existente en la colección db1.coll1 por _id, el campo de coincidencia predeterminado ya que on no está definido explícitamente. Si es así y fullDocument._isDelete está configurado en true, Atlas elimina el documento correspondiente. Si coincide y fullDocument._isDelete está configurado en false, Atlas sustituye el documento coincidente con el nuevo de la fuente de datos en transmisión.

   • En segundo lugar, si Atlas Stream Processing no encuentra ningún documento coincidente y fullDocument._isDelete es verdadero, Atlas descarta el documento en lugar de escribirlo en la colección. Si no existe un documento que coincida de esta manera y fullDocument._isDelete es falso, Atlas inserta el documento de la fuente de datos de transmisión en la colección.

{
  $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”
      }
    },
  }
}