Apache Pulsar 커넥터

Apache Pulsar 커넥터 (Apache Pulsar Connector)

Flink는 exactly-once 보장으로 Pulsar 토픽에서 데이터를 읽고 쓸 수 있는 Apache Pulsar 커넥터를 제공합니다. Pulsar 2.10.0 이상의 버전과 함께 사용할 수 있습니다.

출처: 문서

본문

Flink는 exact-once 보장으로 Pulsar 토픽에서 데이터를 읽고 쓰는 Apache Pulsar 커넥터를 제공합니다.

의존성 (Dependency)

Pulsar 2.10.0 이상에서 커넥터를 사용할 수 있습니다. 항상 최신 Pulsar 버전을 사용하는 것을 권장합니다. Pulsar 호환성에 대한 자세한 내용은 PIP-72에서 확인할 수 있습니다.

Flink 버전 2.3용 커넥터는 아직 없습니다.

PyFlink 작업에서 사용하려면 다음 의존성이 필요합니다:

버전	PyFlink JAR
flink-connector-pulsar	Flink 버전 2.3용 SQL jar는 아직 없습니다.

PyFlink에서 JAR을 사용하는 방법에 대한 자세한 내용은 Python dependency management를 참고하세요.

Flink의 스트리밍 커넥터는 바이너리 배포판의 일부가 아닙니다. 클러스터 실행을 위해 연결하는 방법은 여기에서 확인하세요.

Pulsar 소스 (Pulsar Source)

이 부분은 새 data source API를 기반으로 한 Pulsar 소스를 설명합니다.

사용법 (Usage)

Pulsar 소스는 PulsarSource 인스턴스를 구성하기 위한 빌더 클래스를 제공합니다. 아래 코드 스니펫은 PulsarSource 인스턴스를 빌드합니다. Exclusive 구독 유형(my-subscription)으로 "persistent://public/default/my-topic" 토픽의 가장 이른 커서부터 메시지를 소비하고 메시지의 원시 페이로드를 문자열로 역직렬화합니다.

PulsarSource<String> source = PulsarSource.builder()
    .setServiceUrl(serviceUrl)
    .setAdminUrl(adminUrl)
    .setStartCursor(StartCursor.earliest())
    .setTopics("my-topic")
    .setDeserializationSchema(new SimpleStringSchema())
    .setSubscriptionName("my-subscription")
    .build();
env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");

pulsar_source = PulsarSource.builder() \
    .set_service_url('pulsar://localhost:6650') \
    .set_admin_url('http://localhost:8080') \
    .set_start_cursor(StartCursor.earliest()) \
    .set_topics("my-topic") \
    .set_deserialization_schema(SimpleStringSchema()) \
    .set_subscription_name('my-subscription') \
    .build()

env.from_source(source=pulsar_source,
                watermark_strategy=WatermarkStrategy.for_monotonous_timestamps(),
                source_name="pulsar source")

PulsarSource를 빌드하는 데 다음 속성은 필수입니다:

• Pulsar service URL, setServiceUrl(String)로 구성
• Pulsar service HTTP URL(admin URL이라고도 함), setAdminUrl(String)로 구성
• Pulsar 구독 이름, setSubscriptionName(String)로 구성
• 구독할 토픽/파티션, 아래 topic-partition subscription 참고
• Pulsar 메시지를 파싱할 역직렬화기, 아래 deserializer 참고

소스에 setConsumerName(String)으로 consumer 이름을 설정하는 것을 권장합니다. 이는 Pulsar 통계 대시보드에서 Flink 커넥터에 고유한 이름을 설정합니다. 이를 사용하여 Flink 커넥터와 애플리케이션의 성능을 모니터링할 수 있습니다.

토픽-파티션 구독 (Topic-partition Subscription)

Pulsar 소스는 토픽-파티션 구독의 두 가지 방식을 제공합니다:

• 토픽 목록, 토픽 목록의 모든 파티션에서 메시지 구독. 예:

PulsarSource.builder().setTopics("some-topic1", "some-topic2");
// Partition 0 and 2 of topic "topic-a"
PulsarSource.builder().setTopics("topic-a-partition-0", "topic-a-partition-2");

PulsarSource.builder().set_topics(["some-topic1", "some-topic2"])
# Partition 0 and 2 of topic "topic-a"
PulsarSource.builder().set_topics(["topic-a-partition-0", "topic-a-partition-2"])

• 토픽 패턴, 제공된 정규식과 이름이 일치하는 모든 토픽에서 메시지 구독. 예:

PulsarSource.builder().setTopicPattern("topic-.*");

PulsarSource.builder().set_topic_pattern("topic-.*")

유연한 토픽 이름 지정 (Flexible Topic Naming)

Pulsar 2.0부터 모든 토픽 이름은 내부적으로 {persistent|non-persistent}://tenant/namespace/topic 형태입니다. 이제 파티션된 토픽의 경우 많은 경우에 짧은 이름을 사용할 수 있습니다(단순함을 위해). 유연한 이름 지정 시스템은 Pulsar 클러스터에 이제 기본 토픽 타입, tenant, namespace가 있다는 사실에서 비롯됩니다.

토픽 속성	기본값
topic type	persistent
tenant	public
namespace	default

이 표는 입력 토픽 이름과 번역된 토픽 이름 사이의 매핑 관계를 나열합니다:

입력 토픽 이름	번역된 토픽 이름
my-topic	persistent://public/default/my-topic
my-tenant/my-namespace/my-topic	persistent://my-tenant/my-namespace/my-topic

non-persistent 토픽의 경우 기본 기반 규칙이 non-partitioned 토픽에 적용되지 않으므로 전체 토픽 이름을 지정해야 합니다. 따라서 non-persistent://my-topic 같은 짧은 이름은 사용할 수 없고 대신 non-persistent://public/default/my-topic을 사용해야 합니다.

Pulsar 토픽 파티션 구독 (Subscribing Pulsar Topic Partition)

내부적으로 Pulsar는 파티션 크기에 따라 파티션된 토픽을 non-partitioned 토픽 집합으로 나눕니다.

예를 들어 sample tenant의 flink namespace 아래에 3개 파티션이 있는 simple-string 토픽이 생성되면 Pulsar의 토픽은 다음과 같습니다:

토픽 이름	파티션됨
persistent://sample/flink/simple-string	Y
persistent://sample/flink/simple-string-partition-0	N
persistent://sample/flink/simple-string-partition-1	N
persistent://sample/flink/simple-string-partition-2	N

위의 non-partitioned 토픽 이름을 사용하여 토픽 파티션에서 직접 메시지를 소비할 수 있습니다. 예를 들어 PulsarSource.builder().setTopics("sample/flink/simple-string-partition-1", "sample/flink/simple-string-partition-2")는 sample/flink/simple-string 토픽의 파티션 1과 2를 소비합니다.

토픽 패턴 설정 (Setting Topic Patterns)

Pulsar 소스는 정규식을 사용하여 오직 하나의 tenant와 하나의 namespace 아래의 토픽 집합에 구독할 수 있습니다. 그러나 토픽 타입(persistent 또는 non-persistent)은 정규식으로 결정되지 않습니다. PulsarSource.builder().setTopicPattern("non-persistent://public/default/my-topic.*")을 사용해도 우리는 이름이 public/default/my-topic.*와 일치하는 persistent 및 non-persistent 토픽을 모두 구독합니다.

non-persistent 토픽만 구독하려면 RegexSubscriptionMode를 RegexSubscriptionMode.NonPersistentOnly로 설정해야 합니다. 예를 들어 setTopicPattern("topic-.*", RegexSubscriptionMode.NonPersistentOnly)입니다. setTopicPattern("topic-.*", RegexSubscriptionMode.PersistentOnly)는 persistent 토픽만 구독합니다.

정규식은 토픽 이름 지정 패턴을 따라야 합니다. 토픽 이름 부분만 정규식일 수 있습니다. 예를 들어 some-topic-\d 같은 단순 토픽 정규식을 제공하면 public tenant의 default namespace 아래의 모든 토픽을 필터링합니다. 토픽 정규식이 flink/sample/topic-.*이면 flink tenant의 sample namespace 아래의 모든 토픽을 필터링합니다.

현재 최신 릴리스된 Pulsar 2.11.0은 non-persistent 토픽을 올바르게 반환하지 않았습니다. Pulsar 2.11.0에서는 non-persistent 토픽 필터링에 정규식을 사용할 수 없습니다.

이 버그의 자세한 내용은 이슈 https://github.com/apache/pulsar/issues/19316 참고.

역직렬화기 (Deserializer)

역직렬화기(PulsarDeserializationSchema)는 Pulsar 메시지를 bytes에서 디코딩하는 데 사용됩니다. setDeserializationSchema(PulsarDeserializationSchema)로 구성할 수 있습니다. PulsarDeserializationSchema는 Pulsar Message<byte[]>를 역직렬화하는 방법을 정의합니다.

메시지의 원시 페이로드(바이트 단위 메시지 데이터)만 필요한 경우 미리 정의된 PulsarDeserializationSchema를 사용할 수 있습니다. Pulsar 커넥터는 세 가지 구현 메서드를 제공합니다.

• Pulsar의 Schema로 메시지 디코딩. KeyValue 타입 또는 Struct 타입을 사용하면 pulsar Schema에는 타입 클래스 정보가 없습니다. 하지만 여전히 PulsarSchemaTypeInformation을 구성해야 합니다. 그래서 우리는 타입 정보를 전달하는 두 개의 추가 API를 제공합니다.

// Primitive types
PulsarSourceBuilder.setDeserializationSchema(Schema);
// Struct types (JSON, Protobuf, Avro, etc.)
PulsarSourceBuilder.setDeserializationSchema(Schema, Class);
// KeyValue type
PulsarSourceBuilder.setDeserializationSchema(Schema, Class, Class);

• Flink의 DeserializationSchema로 메시지 디코딩

PulsarSourceBuilder.setDeserializationSchema(DeserializationSchema);

• Flink의 TypeInformation으로 메시지 디코딩

PulsarSourceBuilder.setDeserializationSchema(TypeInformation, ExecutionConfig);

Pulsar Message<byte[]>는 메시지 키, 메시지 게시 시간, 메시지 시간, 애플리케이션 정의 key/value 쌍 같은 일부 추가 속성을 포함합니다. 이러한 속성은 Message<byte[]> 인터페이스에 정의될 수 있습니다.

이 속성들로 Pulsar 메시지를 역직렬화하려면 PulsarDeserializationSchema를 구현해야 합니다. PulsarDeserializationSchema.getProducedType()의 TypeInformation이 올바른지 확인하세요. Flink는 이 TypeInformation을 사용하여 메시지를 다운스트림 연산자로 전달합니다.

소스에서의 스키마 진화 (Schema Evolution in Source)

Schema evolution은 사용자가 Pulsar의 Schema와 PulsarSourceBuilder.enableSchemaEvolution()을 사용하여 활성화할 수 있습니다. 이는 브로커 스키마 검증이 적용된다는 뜻입니다.

Schema<SomePojo> schema = Schema.AVRO(SomePojo.class);
PulsarSource<SomePojo> source = PulsarSource.builder()
    ...
    .setDeserializationSchema(schema, SomePojo.class)
    .enableSchemaEvolution()
    .build();

스키마 진화를 활성화하지 않고 Pulsar 스키마를 사용하면 스키마 검사를 우회합니다. 이는 메시지를 역직렬화할 때 잘못된 스키마를 사용하면 오류가 발생할 수 있습니다.

Auto Consume Schema 사용 (Use Auto Consume Schema)

Pulsar는 미리 정의된 스키마 없이 메시지를 소비하는 Schema.AUTO_CONSUME()를 제공합니다. 이는 토픽이 여러 스키마를 갖고 서로 호환되지 않을 수 있을 때 항상 사용됩니다. Pulsar는 사용자를 위해 메시지를 GenericRecord로 자동 디코딩합니다.

그러나 PulsarSourceBuilder.setDeserializationSchema(Schema) 메서드는 Schema.AUTO_CONSUME()를 지원하지 않습니다. 대신 우리는 GenericRecord를 역직렬화하는 GenericRecordDeserializer를 제공합니다. 이 인터페이스를 구현하여 PulsarSourceBuilder.setDeserializationSchema(GenericRecordDeserializer)에 설정할 수 있습니다.

Pulsar 싱크 (Pulsar Sink)

stream = ...
pulsar_sink = PulsarSink.builder() \
    .set_service_url('pulsar://localhost:6650') \
    .set_admin_url('http://localhost:8080') \
    .set_topics("topic1") \
    .set_serialization_schema(SimpleStringSchema()) \
    .set_delivery_guarantee(DeliveryGuarantee.AT_LEAST_ONCE) \
    .build()
stream.sink_to(pulsar_sink)

PulsarSink을 빌드하는 데 다음 속성은 필수입니다:

• Pulsar service url, setServiceUrl(String)로 구성
• Pulsar service http url(aka. admin url), setAdminUrl(String)로 구성
• 쓸 토픽/파티션, 아래 Producing to topics 참고
• Pulsar 메시지를 생성할 직렬화기, 아래 serializer 참고

소스에 setProducerName(String)으로 producer 이름을 설정하는 것을 권장합니다. 이는 Pulsar 통계 대시보드에서 Flink 커넥터에 고유한 이름을 설정합니다. 이를 사용하여 Flink 커넥터와 애플리케이션의 성능을 모니터링할 수 있습니다.

토픽에 생성 (Producing to topics)

생성할 토픽 정의는 Pulsar 소스의 토픽-파티션 구독과 유사합니다. 토픽 설정의 믹스인 스타일을 지원합니다. 토픽 목록, 파티션, 또는 둘 다를 제공할 수 있습니다.

// Topic "some-topic1" and "some-topic2"
PulsarSink.builder().setTopics("some-topic1", "some-topic2")
// Partition 0 and 2 of topic "topic-a"
PulsarSink.builder().setTopics("topic-a-partition-0", "topic-a-partition-2")
// Partition 0 and 2 of topic "topic-a" and topic "some-topic2"
PulsarSink.builder().setTopics("topic-a-partition-0", "topic-a-partition-2", "some-topic2")

# Topic "some-topic1" and "some-topic2"
PulsarSink.builder().set_topics(["some-topic1", "some-topic2"])
# Partition 0 and 2 of topic "topic-a"
PulsarSink.builder().set_topics(["topic-a-partition-0", "topic-a-partition-2"])
# Partition 0 and 2 of topic "topic-a" and topic "some-topic2"
PulsarSink.builder().set_topics(["topic-a-partition-0", "topic-a-partition-2", "some-topic2"])

제공한 토픽은 자동 파티션 발견을 지원합니다. 고정 간격으로 Pulsar에서 토픽 메타데이터를 조회합니다. PulsarSinkOptions.PULSAR_TOPIC_METADATA_REFRESH_INTERVAL 옵션으로 발견 간격 옵션을 변경할 수 있습니다.

작성 대상을 구성하는 것을 사용자 지정 [TopicRouter] 메시지 라우팅으로 대체할 수 있습니다. Pulsar 커넥터에서 파티션 구성은 유연한 토픽 이름 지정 섹션에서 설명됩니다.

토픽과 해당 파티션을 모두 기반으로 Pulsar 싱크를 빌드하면 Pulsar 싱크는 이를 병합하고 토픽만 사용합니다.

예를 들어 PulsarSink.builder().setTopics("some-topic1", "some-topic1-partition-0") 옵션으로 Pulsar 싱크를 빌드하면 PulsarSink.builder().setTopics("some-topic1")로 단순화됩니다.

토픽은 빌더에서 고정 토픽 집합을 제공하는 대신 들어오는 메시지로 정의할 수 있습니다. 사용자 지정 TopicRouter에서 토픽을 동적으로 제공할 수 있습니다. 토픽 메타데이터는 PulsarSinkContext.topicMetadata(String)로 조회할 수 있으며, 조회 결과는 PulsarSinkOptions.PULSAR_TOPIC_METADATA_REFRESH_INTERVAL 밀리초 후에 캐시되고 만료됩니다.

존재하지 않는 토픽에 쓰고 싶다면 TopicRouter에서 그것을 반환하면 됩니다. Pulsar 커넥터가 그것을 생성하려 시도합니다.

존재하지 않는 토픽에 메시지를 쓰려면 Pulsar의 broker.conf에서 토픽 자동 생성을 활성화해야 합니다. allowAutoTopicCreation=true로 설정하여 활성화하세요.

더 알아보기 (Learn more)

• DataStream 커넥터 개요
• Apache Pulsar
• Table 커넥터 개요