스트림 스키마와 CDC
스트림의 스키마는 열의 자료형, 파티션 키와 CDC(Change Data Capture) 구성을 지정합니다. 스트리밍 데이터셋 생성, 브랜치 스트림 생성과 초기화에서 사용합니다.
최상위 필드
| 필드 | 자료형 | 필수 여부 | 의미 |
|---|---|---|---|
fields | list<Field> | 선택 | 스트림 필드 정의 |
keyFieldNames | list<FieldName> | 선택 | 파티션 키로 사용할 필드 이름 |
changeDataCapture | union<fullRow> | 선택 | 전체 행 기반 CDC 구성 |
FieldName은 문자열입니다. Field에는 필수 문자열 name과 필수 schema 객체가 있습니다. 각 schema 객체는 다음 구조를 사용합니다.
| 필드 | 자료형 | 필수 여부 |
|---|---|---|
nullable | boolean | 필수 |
customMetadata | map<string, any> | 선택 |
dataType | 아래 자료형 중 하나의 객체 | 필수 |
nullable은 null 허용 여부이며 customMetadata는 필드의 추가 메타데이터입니다. 배열 항목, 맵의 키와 값 및 구조체의 하위 필드도 같은 스키마 구조를 재귀적으로 사용합니다.
자료형
dataType.type으로 분기를 선택합니다.
type | 추가 필드 |
|---|---|
boolean | 없음 |
byte | 없음 |
short | 없음 |
integer | 없음 |
long | 없음 |
float | 없음 |
double | 없음 |
decimal | 선택 정수 precision, scale |
string | 없음 |
binary | 없음 |
date | 없음 |
timestamp | 없음 |
struct | 선택 subFields: list<Field> |
array | 필수 itemsSchema |
map | 필수 keySchema, valueSchema |
itemsSchema, keySchema, valueSchema에는 위의 nullable, customMetadata, dataType 구조를 사용합니다. decimal.precision은 전체 자릿수, decimal.scale은 소수점 오른쪽 자릿수이며 각각 최대 38입니다.
파티션 키
keyFieldNames는 같은 키의 레코드를 같은 파티션에 보내 해당 그룹의 처리 순서를 유지하는 데 사용합니다. 고유 레코드를 식별하거나 중복을 제거하는 설정은 아닙니다. 중복 해소가 필요하면 CDC 구성을 함께 지정합니다.
키 필드에는 다음 9가지 자료형을 사용할 수 있습니다.
boolean, byte, date, decimal, integer, long, short, string, timestamp
timestamp 필드를 스트림 키로 사용하는 레코드를 게시할 때는 Unix epoch 밀리초 값을 전달해야 합니다. 키 선택과 전체 처리 경로의 순서 조건은 스트리밍 키를 따릅니다.
전체 행 기반 CDC
CDC를 사용하려면 파티션 키를 하나 이상 지정합니다. changeDataCapture의 허용 분기는 fullRow이며 다음 필드가 모두 필수입니다.
| 필드 | 의미 |
|---|---|
type | fullRow |
deletionFieldName | 행의 삭제 여부를 나타내는 boolean 필드 이름 |
orderingFieldName | 행의 최신 상태를 결정하는 정렬 필드 이름 |
정렬 필드에는 byte, date, decimal, integer, long, short, string, timestamp를 사용할 수 있습니다. 파티션 키와 달리 boolean은 이 목록에 없습니다.
행의 값이 바뀌면 그 행의 전체 새 상태를 게시합니다. 변경한 필드만 보내 이전 값에 합치는 부분 갱신 형식으로 처리하지 않습니다. CDC 소비와 원시 변경 로그의 차이는 변경 데이터 캡처에 있습니다.
다음은 문자열 id를 키로 사용하고 revision으로 최신 상태를 정하는 스키마 예시입니다.
{
"fields": [
{"name":"id","schema":{"nullable":false,"dataType":{"type":"string"}}},
{"name":"value","schema":{"nullable":true,"dataType":{"type":"string"}}},
{"name":"deleted","schema":{"nullable":false,"dataType":{"type":"boolean"}}},
{"name":"revision","schema":{"nullable":false,"dataType":{"type":"long"}}}
],
"keyFieldNames": ["id"],
"changeDataCapture": {
"type":"fullRow",
"deletionFieldName":"deleted",
"orderingFieldName":"revision"
}
}