스키마 관리

스키마 관리 (Manage Schemas)

Pulsar 스키마는 토픽에 저장되는 메시지의 데이터 구조를 정의하고, 클라이언트가 구조를 알고 데이터를 안전하게 직렬화·역직렬화할 수 있게 해줘요. 이 페이지에서는 스키마의 업로드·조회·추출·삭제와 네임스페이스 수준의 자동 업데이트·검증 강제·호환성 전략 설정을 pulsar-admin CLI, REST API, Java admin API로 정리했어요.

출처: 문서

본문

tip

이 페이지는 자주 사용하는 일부 작업만 보여줘요.

• Pulsar admin 명령, 플래그, 설명 등 최신·전체 정보는 Pulsar admin docs를 참고해요.
• REST API 파라미터, 응답, 샘플 등 최신·전체 정보는 REST API doc을 참고해요.
• Java admin API 클래스, 메서드, 설명 등 최신·전체 정보는 Java admin API doc을 참고해요.

스키마 관리 (Manage schema)

스키마 업로드 (Upload a schema)

토픽에 새 스키마를 업로드(등록)하려면 다음 방법 중 하나를 사용할 수 있어요.

pulsar-admin · REST API · Java

upload 하위 명령을 사용해요.

pulsar-admin schemas upload --filename <schema-definition-file> <topic-name>

schema-definition-file은 JSON 형식이에요.

{
    "type": "<schema-type>",
    "schema": "<an-utf8-encoded-string-of-schema-definition-data>",
    "properties": {} // the properties associated with the schema
}

여기에 문서화된 엔드포인트로 POST 요청을 보내요.

REST API: POST /admin/v2/schemas/{tenant}/{namespace}/{topic}/schema

아래는 schema.json 파일에 payload가 저장되어 있고, Pulsar 브로커가 localhost에서 실행 중이며 토픽이 my-tenant/my-ns/my-topic일 때의 CURL 예시예요.

curl -X POST -H 'Content-Type: application/json' -d @schema.json http://localhost:8080/admin/v2/schemas/my-tenant/my-ns/my-topic/schema

POST payload는 JSON 형식이에요.

{
    "type": "<schema-type>",
    "schema": "<an-utf8-encoded-string-of-schema-definition-data>",
    "properties": {} // the properties associated with the schema
}

PulsarAdmin 클라이언트의 메서드는 다음과 같아요.

void createSchema(String topic, PostSchemaPayload schemaPayload)

PostSchemaPayload 예시:

PulsarAdmin admin = …;

PostSchemaPayload payload = new PostSchemaPayload();
payload.setType("INT8");
payload.setSchema("");

admin.createSchema("my-tenant/my-ns/my-topic", payload);

스키마가 primitive(원시) 스키마라면 schema 필드는 비어 있어야 해요. 스키마가 struct(구조) 스키마라면 이 필드는 Avro 스키마 정의의 JSON 문자열이어야 해요.

payload에 포함되는 필드:

Field	Description
type	primitive-type 스키마의 허용 값은 Primitive types 페이지에 나열돼 있어요. struct-type 스키마의 허용 값은 AVRO, PROTOBUF, PROTOBUF_NATIVE, JSON이에요.
schema	스키마 정의 데이터로, UTF-8 문자셋으로 인코딩돼요. 스키마 유형이 AVRO, PROTOBUF 또는 JSON 스키마면 이 필드는 JSON 형식의 Avro 스키마 정의여야 해요. 스키마 유형이 PROTOBUF_NATIVE면 이 필드는 Protobuf 디스크립터를 포함해야 해요. 그 외에는 이 필드가 비어 있어야 해요.
properties	스키마와 연관된 추가 프로퍼티

다음은 JSON 스키마 예시예요.

Example

{
    "type": "JSON",
    "schema": "{\"type\":\"record\",\"name\":\"User\",\"namespace\":\"com.foo\",\"fields\":[{\"name\":\"file1\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"file2\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"file3\",\"type\":[\"string\",\"null\"],\"default\":\"dfdf\"}]}",
    "properties": {}
}

최신 스키마 가져오기 (Get the latest schema)

토픽의 최신 스키마를 가져오려면 다음 방법 중 하나를 사용할 수 있어요.

pulsar-admin · REST API · Java

get 하위 명령을 사용해요.

pulsar-admin schemas get <topic-name>

예시 출력:

{
    "version": 0,
    "type": "String",
    "timestamp": 0,
    "data": "string",
    "properties": {
        "property1": "string",
        "property2": "string"
    }
}

이 엔드포인트로 GET 요청을 보내요.

REST API: GET /admin/v2/schemas/{tenant}/{namespace}/{topic}/schema

JSON 형식으로 반환되는 응답 예시:

{
    "version": "<the-version-number-of-the-schema>",
    "type": "<the-schema-type>",
    "timestamp": "<the-creation-timestamp-of-the-version-of-the-schema>",
    "data": "<an-utf8-encoded-string-of-schema-definition-data>",
    "properties": {} // the properties associated with the schema
}

SchemaInfo createSchema(String topic)

SchemaInfo 예시:

PulsarAdmin admin = …;

SchemaInfo si = admin.getSchema("my-tenant/my-ns/my-topic");

특정 스키마 가져오기 (Get a specific schema)

스키마의 특정 버전을 가져오려면 다음 방법 중 하나를 사용할 수 있어요.

pulsar-admin · REST API · Java

get 하위 명령을 사용해요.

pulsar-admin schemas get <topic-name> --version <version>

스키마 엔드포인트로 GET 요청을 보내요.

REST API: GET /admin/v2/schemas/{tenant}/{namespace}/{topic}/schema/{version}

JSON 형식으로 반환되는 응답 예시:

{
    "version": "<the-version-number-of-the-schema>",
    "type": "<the-schema-type>",
    "timestamp": "<the-creation-timestamp-of-the-version-of-the-schema>",
    "data": "<an-utf8-encoded-string-of-schema-definition-data>",
    "properties": {} // the properties associated with the schema
}

SchemaInfo createSchema(String topic, long version)

SchemaInfo 예시:

PulsarAdmin admin = …;

SchemaInfo si = admin.getSchema("my-tenant/my-ns/my-topic", 1L);

스키마 추출 (Extract a schema)

토픽을 통해 스키마를 추출(제공)하려면 다음 방법을 사용해요.

pulsar-admin

extract 하위 명령을 사용해요.

pulsar-admin schemas extract --classname <class-name> --jar <absolute-jar-path> --type <type-name>

스키마 삭제 (Delete a schema)

note

어떤 경우든 delete 동작은 토픽에 등록된 스키마의 모든 버전을 삭제해요.

토픽의 스키마를 삭제하려면 다음 방법 중 하나를 사용할 수 있어요.

pulsar-admin · REST API · Java

delete 하위 명령을 사용해요.

pulsar-admin schemas delete <topic-name>

스키마 엔드포인트로 DELETE 요청을 보내요.

REST API: DELETE /admin/v2/schemas/{tenant}/{namespace}/{topic}/schema

JSON 형식으로 반환되는 응답 예시:

{
    "version": "<the-latest-version-number-of-the-schema>",
}

void deleteSchema(String topic)

스키마 삭제 예시:

PulsarAdmin admin = …;

admin.deleteSchema("my-tenant/my-ns/my-topic");

스키마 자동 업데이트 관리 (Manage schema AutoUpdate)

스키마 자동 업데이트 활성화 (Enable schema AutoUpdate)

네임스페이스 수준에서 스키마 자동 업데이트를 활성화/강제하려면 다음 방법 중 하나를 사용할 수 있어요.

pulsar-admin · REST API · Java

set-is-allow-auto-update-schema 하위 명령을 사용해요.

bin/pulsar-admin namespaces set-is-allow-auto-update-schema --enable tenant/namespace

네임스페이스 엔드포인트로 POST 요청을 보내요.

REST API: POST /admin/v2/namespaces/{tenant}/{namespace}/isAllowAutoUpdateSchema

POST payload는 JSON 형식이에요.

{
"isAllowAutoUpdateSchema": "true"
}

tenant/namespace의 스키마 자동 업데이트를 활성화하는 예시:

admin.namespaces().setIsAllowAutoUpdateSchema("my-namspace", true);

스키마 자동 업데이트 비활성화 (Disable schema AutoUpdate)

note

스키마 자동 업데이트가 비활성화되면 새 스키마 등록만 할 수 있어요.

네임스페이스 수준에서 스키마 자동 업데이트를 비활성화하려면 다음 명령 중 하나를 사용할 수 있어요.

pulsar-admin · REST API · Java

set-is-allow-auto-update-schema 하위 명령을 사용해요.

bin/pulsar-admin namespaces set-is-allow-auto-update-schema --disable tenant/namespace

네임스페이스 엔드포인트로 POST 요청을 보내요.

REST API: POST /admin/v2/namespaces/{tenant}/{namespace}/isAllowAutoUpdateSchema

POST payload는 JSON 형식이에요.

{
"isAllowAutoUpdateSchema": "false"
}

tenant/namespace의 스키마 자동 업데이트를 비활성화하는 예시:

admin.namespaces().setIsAllowAutoUpdateSchema("my-namspace", false);

스키마 검증 강제 관리 (Manage schema validation enforcement)

스키마 검증 강제 활성화 (Enable schema validation enforcement)

클러스터 수준에서 스키마 검증 강제를 적용하려면 conf/broker.conf 파일에서 isSchemaValidationEnforced를 true로 구성해요.

네임스페이스 수준에서 스키마 검증 강제를 활성화하려면 다음 명령 중 하나를 사용할 수 있어요.

pulsar-admin · REST API · Java

set-schema-validation-enforce 하위 명령을 사용해요.

bin/pulsar-admin namespaces set-schema-validation-enforce --enable tenant/namespace

네임스페이스 엔드포인트로 POST 요청을 보내요.

REST API: POST /admin/v2/namespaces/{tenant}/{namespace}/schemaValidationEnforced

POST payload는 JSON 형식이에요.

{
"schemaValidationEnforced": "true"
}

tenant/namespace의 스키마 검증 강제를 활성화하는 예시:

admin.namespaces().setSchemaValidationEnforced("my-namspace", true);

스키마 검증 강제 비활성화 (Disable schema validation enforcement)

네임스페이스 수준에서 스키마 검증 강제를 비활성화하려면 다음 명령 중 하나를 사용할 수 있어요.

pulsar-admin · REST API · Java

set-schema-validation-enforce 하위 명령을 사용해요.

bin/pulsar-admin namespaces set-schema-validation-enforce --disable tenant/namespace

네임스페이스 엔드포인트로 POST 요청을 보내요.

REST API: POST /admin/v2/namespaces/{tenant}/{namespace}/schemaValidationEnforced

POST payload는 JSON 형식이에요.

{
"schemaValidationEnforced": "false"
}

tenant/namespace의 스키마 검증 강제를 활성화하는 예시:

admin.namespaces().setSchemaValidationEnforced("my-namspace", false);

스키마 호환성 전략 관리 (Manage schema compatibility strategy)

서로 다른 수준에 구성된 스키마 호환성 검사 전략은 우선순위가 있어요: 토픽 수준 > 네임스페이스 수준 > 클러스터 수준. 즉:

• 토픽과 네임스페이스 수준 모두에 전략을 설정하면 토픽 수준 전략이 사용돼요.
• 네임스페이스와 클러스터 수준 모두에 전략을 설정하면 네임스페이스 수준 전략이 사용돼요.

스키마 호환성 전략 설정 (Set schema compatibility strategy)

토픽 수준 스키마 호환성 전략 설정

토픽 수준에서 스키마 호환성 검사 전략을 설정하려면 다음 방법 중 하나를 사용할 수 있어요.

pulsar-admin · REST API · Java

pulsar-admin topicPolicies set-schema-compatibility-strategy 명령을 사용해요.

pulsar-admin topicPolicies set-schema-compatibility-strategy <strategy> <topicName>

이 엔드포인트로 PUT 요청을 보내요.

REST API: PUT /admin/v2/persistent/{tenant}/{namespace}/{topic}/schemaCompatibilityStrategy

void setSchemaCompatibilityStrategy(String topic, SchemaCompatibilityStrategy strategy)

토픽 수준에서 스키마 호환성 검사 전략을 설정하는 예시:

PulsarAdmin admin = …;

admin.topicPolicies().setSchemaCompatibilityStrategy("my-tenant/my-ns/my-topic", SchemaCompatibilityStrategy.ALWAYS_INCOMPATIBLE);

네임스페이스 수준 스키마 호환성 전략 설정

네임스페이스 수준에서 스키마 호환성 검사 전략을 설정하려면 다음 방법 중 하나를 사용할 수 있어요.

pulsar-admin · REST API · Java

pulsar-admin namespaces set-schema-compatibility-strategy 명령을 사용해요.

pulsar-admin namespaces set-schema-compatibility-strategy options

이 엔드포인트로 PUT 요청을 보내요.

REST API: PUT /admin/v2/namespaces/{tenant}/{namespace}/schemaCompatibilityStrategy

setSchemaCompatibilityStrategy 메서드를 사용해요.

admin.namespaces().setSchemaCompatibilityStrategy("test", SchemaCompatibilityStrategy.FULL);

클러스터 수준 스키마 호환성 전략 설정

클러스터 수준에서 스키마 호환성 검사 전략을 설정하려면 conf/broker.conf 파일에서 schemaCompatibilityStrategy를 설정해요.

다음은 예시예요.

schemaCompatibilityStrategy=ALWAYS_INCOMPATIBLE

스키마 호환성 전략 가져오기 (Get schema compatibility strategy)

토픽 수준 스키마 호환성 전략 가져오기

토픽 수준의 스키마 호환성 검사 전략을 가져오려면 다음 방법 중 하나를 사용할 수 있어요.

pulsar-admin · REST API · Java

pulsar-admin topicPolicies get-schema-compatibility-strategy 명령을 사용해요.

pulsar-admin topicPolicies get-schema-compatibility-strategy <topicName>

이 엔드포인트로 GET 요청을 보내요.

REST API: GET /admin/v2/persistent/{tenant}/{namespace}/{topic}/schemaCompatibilityStrategy

SchemaCompatibilityStrategy getSchemaCompatibilityStrategy(String topic, boolean applied)

토픽 수준의 스키마 호환성 검사 전략을 가져오는 예시:

PulsarAdmin admin = …;

// get the current applied schema compatibility strategy
admin.topicPolicies().getSchemaCompatibilityStrategy("my-tenant/my-ns/my-topic", true);

// only get the schema compatibility strategy from topic policies
admin.topicPolicies().getSchemaCompatibilityStrategy("my-tenant/my-ns/my-topic", false);

네임스페이스 수준 스키마 호환성 전략 가져오기

다음 방법 중 하나로 네임스페이스 수준에서 스키마 호환성 검사 전략을 가져올 수 있어요.

pulsar-admin · REST API · Java

pulsar-admin namespaces get-schema-compatibility-strategy 명령을 사용해요.

pulsar-admin namespaces get-schema-compatibility-strategy options

이 엔드포인트로 GET 요청을 보내요.

REST API: GET /admin/v2/namespaces/{tenant}/{namespace}/schemaCompatibilityStrategy

getSchemaCompatibilityStrategy 메서드를 사용해요.

admin.namespaces().getSchemaCompatibilityStrategy("test", SchemaCompatibilityStrategy.FULL);

더 알아보기 (Learn more)

• 스키마 관련 전체 명령은 pulsar-admin의 schemas 명령 참고서를 확인해요.
• REST API 엔드포인트 상세는 /admin/v2/schemas 문서를 참고해요.
• Java admin API의 스키마 메서드는 PulsarAdmin 객체 문서에서 볼 수 있어요.
• primitive-type 스키마와 struct-type 스키마의 개념이 궁금하다면 스키마 개념 문서를 참고해요.