$iceberg etapa de agregación

Conversión de tipo

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

No se admiten otros BSON types. Atlas Stream Processing envía documentos con BSON types no admitidos a la DLQ.

Archivar un Changestream

El siguiente ejemplo muestra cómo guardar el contenido inicial y el changestream de una base de datos de Atlas en una tabla de Apache Iceberg de solo anexar, creando un fichero duradero del historial operativo de esa base de datos. Esta agregación tiene dos etapas:

1. La etapa $source establece una conexión con la base de datos de Atlas, específicamente dirigida a la colección orders en la base de datos db. Permite que la sincronización inicial capture documentos en la base de datos en el momento de la activación del procesador y garantiza que cada evento de flujo de cambios capture el documento completo.

2. La etapa $iceberg establece una conexión con el bucket AWS S3, escribiendo en una tabla denominada myTable en la ruta iceberg-warehouse/. Al especificar solo operaciones insert, garantiza un flujo de escritura de solo anexar y 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"
  }
}

Reflejar una colección

El siguiente ejemplo demuestra cómo reflejar 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 tabla de Apache Iceberg en sincronización con los cambios en la colección de origen de Atlas. Tiene cuatro etapas:

1. La etapa $source establece una conexión con la base de datos de Atlas, específicamente dirigida a la colección orders en la base de datos db. Permite que la sincronización inicial capture documentos en la base de datos en el momento de la activación del procesador y garantiza que cada evento de flujo de cambios capture el documento completo.

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

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

   • Para las operaciones de borrado, cambia la raíz del documento a la clave del documento. Esto da como resultado un registro de que el documento se borró, pero excluye su contenido de un 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 mientras excluye los metadatos de changestream.

4. La etapa $iceberg establece una conexión con el bucket AWS S3, escribiendo en una tabla Apache Iceberg denominada myTable en la ruta iceberg-warehouse/. En el modo cdc, esta etapa determina la operación que se realizará en la tabla Apache Iceberg leyendo del campo de metadatos stream.source.operationType 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"
  }
}

De origen de varias colecciones a destino de varias tablas Apache Iceberg

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

1. La etapa $source establece una conexión con la base de datos de Atlas, específicamente dirigida a las colecciones a, b y c en la base de datos db. Permite que la sincronización inicial capture documentos en la base de datos en el momento de la activación del procesador y garantiza que cada evento de flujo de cambios capture el documento completo.

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

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

   • Para las operaciones de borrado, cambia la raíz del documento a la clave del documento. Esto da como resultado un registro de que el documento se borró, pero excluye su contenido de un 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 mientras excluye los metadatos de changestream.

4. La etapa $iceberg establece una conexión con el bucket AWS S3 denominado myData, escribiendo en una tabla Apache Iceberg en la ruta iceberg-warehouse/. Determina el nombre de la tabla según el nombre del origen collection tal como se recupera 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"
  }
}