$iceberg 애그리게이션 단계

유형 변환

Atlas Stream Processing은 $iceberg 단계의 테이블에 기록할 때 BSON 에서 Iceberg 기본 유형 으로 유형 변환을 수행합니다.

다른 BSON types는 지원되지 않습니다. Atlas Stream Processing은 지원되지 않는 BSON types을 포함한 문서를 DLQ로 보냅니다.

변경 스트림 보관

다음 예시는 Atlas 데이터베이스의 초기 내용과 변경 스트림을 추가 전용 방식으로 Apache Iceberg 테이블에 쓰는 방법을 보여주며, 해당 데이터베이스의 운영 기록의 지속형 보관을 생성합니다. 이 집계는 두 단계로 구성됩니다.

1. $source 단계는 Atlas 데이터베이스와 연결하고 구체적으로 db 데이터베이스의 orders 컬렉션을 대상으로 지정합니다. 이를 통해 초기 동기화를 실행하여 프로세서 활성화 시점에 데이터베이스의 문서를 캡처하고 각 변경 스트림 이벤트로 전체 문서가 캡처되도록 합니다.

2. $iceberg 단계는 AWS S3 버킷에 연결하고 iceberg-warehouse/ 경로의 myTable 라는 이름의 테이블에 쓰기를 수행합니다. insert 작업만 지정하여 추가만 가능한 로그 스타일 쓰기 플로우를 보장합니다.

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

컬렉션 미러링

다음 예시는 Atlas 컬렉션 전체를 Apache Iceberg 테이블로 반영하는 방법을 보여줍니다.

집계를 정의하기 전에 다음 변수를 설정합니다.

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

다음 집계는 Atlas 소스 컬렉션의 변경 사항과 동기화되어 Apache Iceberg 테이블 엔트리를 추가, 업데이트 및 삭제합니다. 이것에는 4단계가 있습니다.

1. $source 단계는 Atlas 데이터베이스와 연결하고 구체적으로 db 데이터베이스의 orders 컬렉션을 대상으로 지정합니다. 이를 통해 초기 동기화를 실행하여 프로세서 활성화 시점에 데이터베이스의 문서를 캡처하고 각 변경 스트림 이벤트로 전체 문서가 캡처되도록 합니다.

2. $match 단계는 operationType 에 대한 필터링을 수행하여 유효한 작업 유형 선언이 포함된 문서만 처리합니다.

3. $replaceRoot 단계는 작업 유형에 따라 문서 루트를 변경합니다.

   • 삭제 작업의 경우 문서 루트를 문서의 키로 변경합니다. 이로 인해 문서가 삭제되었음을 기록하지만 그 내용은 추가 처리에서 제외됩니다.

   • 다른 모든 작업의 경우 문서 루트를 fullDocument로 변경하고 문서 내용을 전달하여 추가 처리하도록 하는 동안 변경 스트림 메타데이터는 제외합니다.

4. $iceberg 단계는 Amazon Web Services S3 버킷에 연결하여 iceberg-warehouse/ 경로의 myTable 이름의 Apache Iceberg 테이블에 기록합니다. cdc 모드에서는 이 단계가 각 문서의 stream.source.operationType 메타데이터 필드에서 읽어 Apache Iceberg 테이블에 대해 수행할 작업을 결정합니다.

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

다중 컬렉션 소스에서 다중 테이블 Apache Iceberg 대상

다음 예시에서 Atlas Stream Processing은 동적 표현식을 사용하여 문서를 다양한 출력 대상으로 동적으로 라우팅합니다.

1. $source 단계는 Atlas 데이터베이스와 연결을 수립하며, 구체적으로 db 데이터베이스의 a, b, c 컬렉션을 대상으로 합니다. 이를 통해 초기 동기화를 실행하여 프로세서 활성화 시점에 데이터베이스의 문서를 캡처하고 각 변경 스트림 이벤트로 전체 문서가 캡처되도록 합니다.

2. $match 단계는 operationType 이 "insert", "update", "delete" 또는 "replace"인 문서를 필터링합니다.

3. $replaceRoot 단계는 작업 유형에 따라 문서 루트를 변경합니다.

   • 삭제 작업의 경우 문서 루트를 문서의 키로 변경합니다. 이로 인해 문서가 삭제되었음을 기록하지만 그 내용은 추가 처리에서 제외됩니다.

   • 다른 모든 작업의 경우 문서 루트를 fullDocument로 변경하고 문서 내용을 전달하여 추가 처리하도록 하는 동안 변경 스트림 메타데이터는 제외합니다.

4. $iceberg 단계는 myData라는 이름의 Amazon Web Services S3 버킷에 연결을 설정하고 iceberg-warehouse/ 경로의 Apache Iceberg 테이블에 쓹니다. 문서 메타데이터에서 조회된 원본 collection 의 이름에 따라 테이블 이름을 결정합니다. 또한 문서 메타데이터에 따라 수행할 작업을 결정합니다.

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