$iceberg etapa de agregación

Conversión de tipo

Atlas Stream Processing realiza una conversión de tipos de BSON a tipos primitivos Iceberg al escribir en tablas en la $iceberg etapa.

Otros tipos BSON no son compatibles. Atlas Stream Processing envía los documentos con tipos BSON no compatibles a la cola de documentos no entregados (DLQ).

Archivar un flujo de cambios

El siguiente ejemplo demuestra cómo escribir el contenido inicial y el flujo de cambios de una base de datos Atlas en una tabla de Apache Iceberg de forma de solo escritura, creando un archivo duradero del historial operativo de esa base de datos. Esta agregación tiene dos etapas:

1. La etapa establece una conexión con la base de datos Atlas, específicamente con $source la orders colección en la base de datos. Permite la db sincronización inicial para capturar documentos en la base de datos al activarse el procesador y garantiza que cada evento de cambio capture el documento completo.

2. La $iceberg etapa establece una conexión con el bucket S3 de AWS, escribiendo en una tabla llamada myTable en la iceberg-warehouse/ ruta. Al especificar solo insert operaciones, garantiza un flujo de escritura de solo adición, estilo registro.

{
  "$source": {
    "connectionName": "atlas1",
    "db": "db",
    "coll": "orders",
    "initialSync": {
      "enable": true
    },
    "config": {
      "fullDocument": "required"
    }
  },
  "$iceberg": {
    "connectionName": "myS3Connection",
    "bucket": "myData",
    "path": "iceberg-warehouse/",
    "tableName": "myTable",
    "mode": "insert"
  }
}

Reflejando una colección

El siguiente ejemplo muestra cómo replicar una colección de Atlas en su totalidad en una tabla de Apache Iceberg.

Antes de definir la agregación, establezca la siguiente variable:

const isDeleteExpr = {$eq: [{$meta: "stream.source.operationType"}, "delete"]};

La siguiente agregación agrega, actualiza y elimina entradas de la tabla Apache Iceberg en sincronía con los cambios en la colección de origen de Atlas. Consta de cuatro etapas:

1. La etapa establece una conexión con la base de datos Atlas, específicamente con $source la orders colección en la base de datos. Permite la db sincronización inicial para capturar documentos en la base de datos al activarse el procesador y garantiza que cada evento de cambio capture el documento completo.

2. La etapa filtra en $match de operationType modo que solo se procesan los documentos con declaraciones de tipo de operación válidas.

3. La etapa cambia la raíz del documento dependiendo del tipo de $replaceRoot operación.

   • Para las operaciones de eliminación, se cambia la raíz del documento a la clave del documento. Esto genera un registro de que el documento fue eliminado, pero excluye su contenido de cualquier procesamiento posterior.

   • Para todas las demás operaciones, cambia la raíz del documento a fullDocument, pasando el contenido del documento para su posterior procesamiento, pero excluyendo los metadatos del flujo de cambios.

4. La $iceberg etapa establece una conexión con el bucket S3 de AWS, escribiendo en una tabla de Apache Iceberg llamada myTable en la iceberg-warehouse/ ruta. En cdc el modo, esta etapa determina la operación a realizar sobre la tabla de Apache Iceberg leyendo el stream.source.operationType campo de metadatos de cada documento.

{
  "$source": {
    "connectionName": "atlas1",
    "db": "db",
    "coll": "orders",
    "initialSync": {
      "enable": true
    },
    "config": {
      "fullDocument": "required"
    }
  },
  "$match": {
    "operationType": {
      "$in": ["insert", "update", "delete", "replace"]
    }
  },
  "$replaceRoot": {
    "newRoot": {
      "$cond": {
        "if": isDeleteExpr,
        "then": "$documentKey",
        "else": "$fullDocument"
      }
    }
  }
  "$iceberg": {
    "connectionName": "myS3Connection",
    "bucket": "myData",
    "path": "iceberg-warehouse/",
    "tableName": "myTable",
    "mode": "cdc"
  }
}

Desde una fuente de múltiples colecciones hasta un objetivo Apache Iceberg de múltiples tablas

En el siguiente ejemplo, Atlas Stream Processing utiliza expresiones dinámicas para enrutar documentos de forma dinámica a diversos destinos de salida.

1. La etapa establece una conexión con la base de datos Atlas, específicamente con las colecciones, y de la base de datos.$source Permite a b c la db sincronización inicial para capturar documentos en la base de datos al activarse el procesador y garantiza que cada evento de cambio capture el documento completo.

2. La etapa $match filtra los documentos cuyo operationType es uno de "insert", "update", "delete" o "replace".

3. La etapa cambia la raíz del documento dependiendo del tipo de $replaceRoot operación.

   • Para las operaciones de eliminación, se cambia la raíz del documento a la clave del documento. Esto genera un registro de que el documento fue eliminado, pero excluye su contenido de cualquier procesamiento posterior.

   • Para todas las demás operaciones, cambia la raíz del documento a fullDocument, pasando el contenido del documento para su posterior procesamiento, pero excluyendo los metadatos del flujo de cambios.

4. La $iceberg etapa establece una conexión con el bucket de AWS S3 myData llamado, escribiendo en una tabla de Apache Iceberg en la iceberg-warehouse/ ruta. Determina el nombre de la tabla según el nombre de la fuente,collection obtenido de los metadatos del documento. También determina la operación a realizar según los metadatos del documento.

{
  "$source": {
    "connectionName": "atlas1",
    "db": "db",
    "coll": ["a", "b", "c"],
    "initialSync": {
      "enable": true
    },
    "config": {
      "fullDocument": "required"
    }
  }
},
{
  "$match": {
    "operationType": {
      "$in": ["insert", "update", "delete", "replace"]
    }
  }
},
{
  "$replaceRoot": {
    "newRoot": {
      "$cond": {
        "if": {
          "$eq": [{ "$meta": "stream.source.operationType" }, "delete"]
        },
        "then": "$documentKey",
        "else": "$fullDocument"
      }
    }
  }
},
{
  "$iceberg": {
    "connectionName": "myS3Connection",
    "databaseName": "iceberg-db",
    "bucket": "myData",
    "path": "iceberg-warehouse/",
    "tableName": { "$meta": "stream.source.ns.coll" },
    "mode": "cdc"
  }
}