$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.

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

Campos de objetos y matrices

El campo schemaInference.mode determina cómo Atlas Stream Processing escribe los campos object y array en las tablas de Apache Iceberg:

• En el modo json, Atlas Stream Processing serializa los campos object y array como cadenas JSON básicas y los escribe en las columnas string. El modo predeterminado es json.

• En el modo nested, Atlas Stream Processing infiere los tipos anidados de Iceberg a partir de los campos object y array:

  • Cada campo object se convierte en una columna struct. Atlas Stream Processing infiere el tipo de cada campo en la estructura a partir del campo correspondiente en el objeto. Esto se aplica en todos los niveles de anidamiento.

  • Cada campo array se convierte en una columna list. Atlas Stream Processing infiere el tipo de elemento de la lista a partir de los elementos no nulos del array. Todos los elementos primitivos deben ser del mismo tipo. Si los elementos son objetos, el tipo de elemento es un struct que contiene los campos de todos los objetos.

Este modo se aplica únicamente a los campos que no tienen una columna coincidente en la tabla de destino. Si la columna ya existe, Atlas Stream Processing ignora el modo y escribe el campo según el tipo de columna.

• Si la columna es string, Atlas Stream Processing serializa el campo como una cadena JSON básica.

• Si la columna es de tipo struct o list, Atlas Stream Processing escribe el campo como ese tipo anidado.

Si la entrada contiene nuevos campos object o array, Atlas Stream Processing utiliza el modo para inferir los tipos de las nuevas columnas.

Archivar un Changestream

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 $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 S3 de AWS, escribiendo en una tabla llamada myTable en la ruta iceberg-warehouse/. Al especificar solo operaciones insert, 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"
  }
}

Reflejar 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 $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 S3 de AWS, escribiendo en una tabla de Apache Iceberg llamada myTable en la ruta iceberg-warehouse/. En el modo cdc, esta etapa determina la operación a realizar sobre la tabla de Apache Iceberg leyendo el 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"
  }
}

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 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 S3 de AWS denominado myData, escribiendo en una tabla de Apache Iceberg en la ruta iceberg-warehouse/. Determina el nombre de la tabla según el nombre de la fuente collection, obtenido de los metadatos del documento. Asimismo, 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"
  }
}