& Flink그 중 다음 두 강의에 대한 내용이다. 강의 중간 중간의 핸즈온도 설명한다:
•
Key Concepts of a Schema Registry
•
The Schema Workflow
Key Concepts of a Schema Registry
스키마의 필요성
카프카 애플리케이션은 크게 프로듀서, 컨슈머로 나뉜다. 둘 사이엔 암묵적인 계약(contract)이 있다. 프로듀서는 “컨슈머가 처리할 수 있는” 이벤트를 발행한다는 것인다. 만약에 이벤트 형식이 바뀐다면?
컨슈머가 바뀐 이벤트를 처리하지 못할 것이다. 그러나 프로듀서는 변할 수 있다. 이럴 경우를 위해 스키마라는 계약이 필요하다.
스키마는 메시지 형식에 대한 규칙이다. 구조, 필드 이름, 필드 타입 등의 정보가 있다. 실제로 아브로나 프로토버프 또는 json 같은 직렬화라고 보아도 된다.
스키마 레지스트리의 필요성
스키마라는 계약서를 프로듀서와 컨슈머 모두 갖고 이행할 수 있도록 중재자이자 변호사로서 스키마 레지스트리가 존재한다.
위 메타포로 이해하자면… 대충 서로 나중에 딴소리 하면 안된다 정도 이겠지만, 기본적으로 프로듀서는 컨슈머가 처리할 수 있는 메시지 형식으로 발행해야 한다는 것. 그리고 나중에 설명할, 데이터가 바뀌기 위해서 스키마 진화 관점에서 스키마 버전 간 호환성이 중요하다.
스키마 레지스트리는 스키마와 스키마 진화(evolve)에 따른 버전 등을 기록하고 있다. 프로듀서와 컨슈머는 여기의 스키마를 활용하여 메시지에 대한 직렬화/역직렬화를 한다.
위의 (컨플루언트) 스키마 레지스트리의 특성을 설명과 함께 코드 예제를 통해 배운다. 실습 환경의 보일러플레이트 코드는 아래 레포에 있다:
kafka streams 101 강의에 비하면 설명이나 readme가 좀 부실하지만 기본적으로 방법은 비슷하다. 하지만 여기선 properties를 만들고 빌드하더라도 실패한다. 처음 제공하는 코드가 완벽하지 않기 때문이다.
따라서 핸즈온이 아닌 강의에서 설명하는 코드 스니펫을 채워가며 빌드, 실행해보면 된다. (컨플루언트 클러스터 생성이나 구성, 프로젝트의 properties 구성은 생략하고 후에 kafka streams 101 포스트에서 설명하겠습니다)
The Schema Workflow(How to Write, Manage and Register Schemas)
스키마 파일 작성
여기선 프로트버프, 아브로 순서로 각각 설명한다:
// 프로트버프
// src/main/proto/purchase.proto
syntax = "proto3";
package io.confluent.developer.proto;
option java_outer_classname = "PurchaseProto";
message Purchase {
string item = 1;
double total_cost = 2;
string customer_id = 3;
}
Protobuf
복사
•
syntax : 프로토버프 버전 선언
•
package : 네임스페이스, 생성할 자바 클래스의 패키지
•
option
◦
java_outer_classname : 생성할 자바 파일 이름
•
message : 생성할 파일의 내부 클래스에 스키마 정의
// 아브로
// src/main/avro/purchase.avsc
{
"type": "record",
"namespace": "io.confluent.developer.avro",
"name": "Purchase",
"fields": [
{
"name": "item",
"type": "string"
},
{
"name": "total_cost",
"type": "double"
},
{
"name": "customer_id",
"type": "string"
}
]
}
JSON
복사
자바 애플리케이션 프로젝트에 스키마 파일 추가
각각 src/main/protobuf와 src/main/avro에 위치시킨다.
스키마 레지스트리 플러그인 구성
plugins {
id "com.google.protobuf" version "0.8.19"
id "com.github.imflog.kafka-schema-registry-gradle-plugin" version "1.6.0"
id "com.github.davidmc24.gradle.plugin.avro" version "1.3.0"
}
Groovy
복사
// See https://github.com/ImFlog/schema-registry-plugin for more details on configuring the Schema Registry plugin
schemaRegistry {
def props = new Properties()
file("src/main/resources/confluent.properties").withInputStream { props.load(it) }
def srUrl = props.getProperty("schema.registry.url")
def fullAuth = props.getProperty("basic.auth.user.info")
if (srUrl != null && fullAuth != null) {
// println "Using Schema Registry endpoint:${srUrl}, username:${auth[0]},password:${auth[1]}"
def auth = fullAuth.split(":")
url = srUrl
credentials {
// username is the characters up to the ':' in the basic.auth.user.info property
username = auth[0]
// password is everything after ':' in the basic.auth.user.info property
password = auth[1]
}
} else {
println("Expected to find the [schema.registry.url] and [basic.auth.user.info]")
}
register {
subject('avro-purchase-value', 'src/main/avro/purchase.avsc', 'AVRO')
subject('proto-purchase-value', 'src/main/proto/purchase.proto', 'PROTOBUF')
}
download {
// commented out to prevent its download which results in the schema
// definition json being flattened to a single line which doesn't
// match the exercise illustration
// subject('avro-purchase-value', 'src/main/avro', 'purchase')
subject('proto-purchase-value', 'src/main/proto', 'purchase')
}
compatibility {
subject('avro-purchase-value', 'src/main/avro/purchase.avsc', 'AVRO')
subject('proto-purchase-value', 'src/main/proto/purchase.proto', 'PROTOBUF')
}
}
Groovy
복사
프로트버프 플러그인 구성
protobuf {
generatedFilesBaseDir = "${project.buildDir}/generated-main-proto-java"
protoc {
artifact = 'com.google.protobuf:protoc:3.18.2'
}
}
Groovy
복사
스키마 모델 객체 생성
./gradlew clean build 로 아브로와 프로트버프 각각 생성된 자바 클래스 파일을 확인한다.
build/generated-main-avro-java/io/confluent/developer/avro/Purchase.java
build/generated-main-proto-java/main/java/io/confluent/developer/proto/PurchaseProto.java