$iceberg 애그리게이션 단계

유형 변환

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

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

변경 스트림 보관

다음 예시 추가 전용 방식으로 Atlas 데이터베이스 의 초기 콘텐츠와 변경 스트림을 Apache ice버그 테이블에 쓰기 (write) 해당 데이터베이스의 운영 기록에 대한 지속형 아카이브를 생성하는 방법을 보여줍니다. 이 집계 에는 두 단계가 있습니다.

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

2. $iceberg 단계는 AWS S3 버킷에 대한 연결을 설정하여 iceberg-warehouse/ 경로에 있는 myTable이라는 이름의 테이블에 기록합니다. insert 연산만 지정하여 추가 전용 로그 스타일 쓰기 (write) 흐름을 보장합니다.

{
  "$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 ice버그 테이블에 미러링하는 방법을 보여줍니다.

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

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 단계는 AWS S3 버킷에 대한 연결을 설정하여 iceberg-warehouse/ 경로에 있는 myTable라는 이름의 Apache 아이스버그 테이블에 기록합니다. 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 아이스버그 대상으로

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

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

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

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

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

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

4. $iceberg 단계는 myData이라는 이름의 AWS S3 버킷에 대한 연결을 설정하여 iceberg-warehouse/ 경로의 Apache 아이스버그 테이블에 기록합니다. 문서 메타데이터 에서 검색된 소스 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"
  }
}