스키마로 작업하기

스키마로 작업하기 (Work with schema)

Pulsar 스키마를 사용하면 토픽에 들어오는 메시지의 데이터 타입과 형식을 보장할 수 있어요. 이 문서는 파이썬 클라이언트로 스키마를 다루는 방법에 초점을 맞춰요. 스키마 개요와 언어별 코드 예시가 필요하면 별도 문서를 참고할 수 있어요.

출처: 문서

본문

스키마 시작하기 (Get started with schema)

Pulsar 스키마의 개요와 언어별 코드 예시는 Schema - Overview 및 Schema - Get Started를 참고하세요.

파이썬 스키마로 작업하기 (Work with Python schema)

파이썬 스키마로 작업하는 것은 다른 언어를 사용하는 것과 약간 다르답니다. 이 섹션에서는 파이썬 클라이언트로 스키마를 작업하는 방법에 대한 구체적인 참조와 예시를 소개해요.

지원되는 스키마 타입 (Supported schema types)

Pulsar에는 다양한 내장 스키마 타입을 사용할 수 있어요. 모든 정의는 pulsar.schema 패키지에 있어요.

스키마 참고
BytesSchema 원시 페이로드를 bytes 객체로 가져와요. 직렬화/역직렬화를 수행하지 않아요. 이게 기본 스키마 모드예요.
StringSchema 페이로드를 UTF-8 문자열로 인코딩/디코딩해요. str 객체를 사용해요.
JsonSchema 레코드 정의가 필요해요. 레코드를 표준 JSON 페이로드로 직렬화해요.
AvroSchema 레코드 정의가 필요해요. AVRO 형식으로 직렬화해요.

스키마 정의 참조 (Schema definition reference)

스키마 정의는 pulsar.schema.Record를 상속하는 클래스를 통해 이뤄져요. 이 클래스에는 pulsar.schema.Field 타입이거나 또 다른 중첩 Record일 수 있는 여러 필드가 있어요. 모든 필드는 pulsar.schema 패키지에 명시되어 있고, 필드는 AVRO 필드 타입과 일치해요.

필드 타입 Python 타입 참고
Boolean bool
Integer int
Long int
Float float
Double float
Bytes bytes
String str
Array list 항목의 레코드 타입을 지정해야 해요.
Map dict 키는 항상 String이에요. 값 타입을 지정해야 해요.

추가로, 모든 Python Enum 타입을 유효한 필드 타입으로 사용할 수 있어요.

필드 파라미터 (Fields parameters)

필드를 추가할 때 생성자에서 다음 파라미터를 사용할 수 있어요.

인자 기본값 참고
default None 필드의 기본값을 설정해요. 예: a = Integer(default=5).
required False 필드를 "required"로 표시해요. 스키마에 그에 맞게 설정돼요.

스키마 정의 예시 (Schema definition examples)

간단한 정의 (Simple definition)

class Example(Record):
    a = String()
    b = Integer()
    c = Array(String())
    i = Map(String())

Enum 사용 (Using enums)

from enum import Enum

class Color(Enum):
    red = 1
    green = 2
    blue = 3

class Example(Record):
    name = String()
    color = Color

복합 타입 (Complex types)

class MySubRecord(Record):
    x = Integer()
    y = Long()
    z = String()

class Example(Record):
    a = String()
    sub = MySubRecord()

Avro 스키마의 네임스페이스 설정 (Set namespace for Avro schema)

특수 필드 _avro_namespace를 사용해 Avro Record 스키마의 네임스페이스를 설정해요.

class NamespaceDemo(Record):
   _avro_namespace = 'xxx.xxx.xxx'
   x = String()
   y = Integer()

스키마 정의는 다음과 같아요.

{
  "name": "NamespaceDemo",
  "namespace": "xxx.xxx.xxx",
  "type": "record",
  "fields": [
    {"name": "x", "type": ["null", "string"]},
    {"name": "y", "type": ["null", "int"]}
  ]
}

스키마 선언 및 검증 (Declare and validate schema)

프로듀서가 만들어지기 전에 Pulsar 브로커는 기존 토픽 스키마가 올바른 타입이고 그 형식이 클래스의 스키마 정의와 호환되는지 검증해요. 토픽 스키마의 형식이 스키마 정의와 호환되지 않으면 프로듀서 생성에서 예외가 발생해요.

특정 스키마 정의로 프로듀서가 한 번 만들어지면, 선언된 스키마 클래스의 인스턴스인 객체만 받아들여요.

마찬가지로, 컨슈머나 리더의 경우 컨슈머는 원시 바이트가 아니라 스키마 레코드 클래스의 인스턴스인 객체를 반환해요.

예시:

consumer = client.subscribe(
                  topic='my-topic',
                  subscription_name='my-subscription',
                  schema=AvroSchema(Example) )

while True:
    msg = consumer.receive()
    ex = msg.value()
    try:
        print("Received message a={} b={} c={}".format(ex.a, ex.b, ex.c))
        # Acknowledge successful processing of the message
        consumer.acknowledge(msg)
    except Exception:
        # Message failed to be processed
        consumer.negative_acknowledge(msg)

더 많은 코드 예시는 Schema - Get started를 참고하세요.

더 알아보기 (Learn more)