$iceberg 애그리게이션 단계

유형 변환

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

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

객체 및 배열 필드

schemaInference.mode 필드 Atlas Stream Processing object 및 array 필드를 Apache iceberg 테이블에 쓰는 방법을 결정합니다.

• json 모드 에서, Atlas Stream Processing object 및 array 필드를 기본 JSON 문자열로 직렬화하여 string 열에 씁니다. json 가 기본값 모드 입니다.

• nested 모드 에서, Atlas Stream Processing object 및 array 필드에서 빙산 중첩 유형을 추론합니다:

  • 각 object 필드 struct 열이 됩니다. Atlas Stream Processing 객체 의 해당 필드 에서 구조체의 각 필드 유형을 추론합니다. 이는 모든 중첩 수준에 적용됩니다.

  • 각 array 필드 list 열이 됩니다. Atlas Stream Processing 배열 의 null이 아닌 요소에서 목록 요소 유형을 추론합니다. 기본 요소는 모두 동일한 유형이어야 합니다. 요소가 객체인 경우 요소 유형은 모든 객체의 필드를 포함하는 struct입니다.

이 모드 대상 테이블에 일치하는 열이 없는 필드에만 적용됩니다. 열이 이미 존재하는 경우 Atlas Stream Processing 모드 무시하고 열 유형에 따라 필드 씁니다.

• 열이 string인 경우 Atlas Stream Processing 필드 기본 JSON 문자열로 직렬화합니다.

• 열이 struct 또는 list인 경우 Atlas Stream Processing 필드 해당 중첩 유형으로 씁니다.

입력에 새 object 또는 array 필드가 포함된 경우, Atlas Stream Processing 모드 를 사용하여 새 열의 유형을 추론합니다.

변경 스트림 보관

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

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 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 단계는 AWS S3 버킷에 대한 연결을 설정하여 iceberg-warehouse/ 경로에 있는 myTable이라는 이름의 Apache 아이스버그 테이블에 기록합니다. cdc 모드 에서, 이 단계에서는 각 문서 의 stream.source.operationType 메타데이터 필드 읽어 Apache ice버그 테이블에 대해 수행할 작업을 결정합니다.

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