KoP 架構介紹

KoP 是 Kafka on Pulsar 的簡寫，顧名思義就是如何在 Pulsar 上實現對 Kafka 數據的讀寫。KoP 將 Kafka 協議處理插件引入 Pulsar Broker 來實現 Apache Pulsar 對 Apache Kafka 協議的支持。將 KoP 協議處理插件添加到現有 Pulsar 集群后，用戶不用修改代碼就可以將現有的 Kafka 應用程序和服務遷移到 Pulsar。

Apache Pulsar 主要特點如下：

利用企業級多租戶特性簡化運營。

避免數據搬遷，簡化操作。

利用 Apache BookKeeper 和分層存儲持久保留事件流。

利用 Pulsar Functions 進行無服務器化事件處理。

KoP 架構如下圖，通過圖可以看到 KoP 引入一個新的協議處理插件，該協議處理插件利用 Pulsar 的現有組件（例如 Topic 發現、分布式日志庫-ManagedLedger、cursor 等）來實現 Kafka 傳輸協議。

Routine Load 訂閱 Pulsar 數據思路

Apache Doris Routine Load 支持了將 Kafka 數據接入 Apache Doris，并保障了數據接入過程中的事務性操作。Apache Pulsar 定位為一個云原生時代企業級的消息發布和訂閱系統，已經在很多線上服務使用。那么 Apache Pulsar 用戶如何將數據接入 Apache Doris 呢，答案是通過 KoP 實現。

由于 KoP 直接在 Pulsar 側提供了對 Kafka 的兼容，那么對于 Apache Doris 來說可以像使用 Kafka 一樣使用 Plusar。整個過程對于 Apache Doris 來說無需任務改變，就能將 Pulsar 數據接入 Apache Doris，并且可以獲得 Routine Load 的事務性保障。

--------------------------

| Apache Doris |

| --------------- |

| | Routine Load | |

| --------------- |

--------------------------

|Kafka Protocol（librdkafka）

------------v--------------

| --------------- |

| | KoP | |

| --------------- |

| Apache Pulsar |

--------------------------

操作實踐

Pulsar Standalone 安裝環境準備：

JDK 安裝：略

下載 Pulsar 二進制包，并解壓：

#下載

wget https://archive.apache.org/dist/pulsar/pulsar-2.10.0/apache-pulsar-2.10.0-bin.tar.gz

#解壓并進入安裝目錄

tar xvfz apache-pulsar-2.10.0-bin.tar.gz

cd apache-pulsar-2.10.0

組件編譯和安裝

1. 下載 KoP 源碼

git clone https://github.com/streamnative/kop.git

cd kop

2. 編譯 KoP 項目

mvn clean install -DskipTests

3. protocols 配置：在解壓后的 apache-pulsar 目錄下創建 protocols文 件夾，并把編譯好的 nar 包復制到 protocols 文件夾中。

mkdir apache-pulsar-2.10.0/protocols

# mv kop/kafka-impl/target/pulsar-protocol-handler-kafka-{{protocol:version}}.nar apache-pulsar-2.10.0/protocols

cp kop/kafka-impl/target/pulsar-protocol-handler-kafka-2.11.0-SNAPSHOT.nar apache-pulsar-2.10.0/protocols

4. 添加后的結果查看

［root@17a5da45700b apache-pulsar-2.10.0］# ls protocols/

pulsar-protocol-handler-kafka-2.11.0-SNAPSHOT.nar

KoP 配置添加

1. 在 standalone.conf 或者 broker.conf 添加如下配置

#kop適配的協議

messagingProtocols=kafka

#kop 的NAR文件路徑

protocolHandlerDirectory=。/protocols

#是否允許自動創建topic

allowAutoTopicCreationType=partitioned

2. 添加如下服務監聽配置

# Use `kafkaListeners` here for KoP 2.8.0 because `listeners` is marked as deprecated from KoP 2.8.0

kafkaListeners=PLAINTEXT://127.0.0.1:9092# This config is not required unless you want to expose another address to the Kafka client.

# If it’s not configured， it will be the same with `kafkaListeners` config by default

kafkaAdvertisedListeners=PLAINTEXT://127.0.0.1:9092

brokerEntryMetadataInterceptors=org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptor

brokerDeleteInactiveTopicsEnabled=false

當出現如下錯誤：

java.lang.IllegalArgumentException： Broker has disabled transaction coordinator， please enable it before using transaction.

添加如下配置，開啟 transactionCoordinatorEnabled

kafkaTransactionCoordinatorEnabled=true

transactionCoordinatorEnabled=true

Pulsar 啟動

#前臺啟動

#bin/pulsar standalone

#后臺啟動

pulsar-daemon start standalone

創建 Doris 數據庫和建表

#進入Doris

mysql -u root -h 127.0.0.1 -P 9030

# 創建數據庫

create database pulsar_doris;

#切換數據庫

use pulsar_doris;

#創建clicklog表

CREATE TABLE IF NOT EXISTS pulsar_doris.clicklog

（

`clickTime` DATETIME NOT NULL COMMENT “點擊時間”，

`type` String NOT NULL COMMENT “點擊類型”，

`id` VARCHAR（100） COMMENT “唯一id”，

`user` VARCHAR（100） COMMENT “用戶名稱”，

`city` VARCHAR（50） COMMENT “所在城市”

）

DUPLICATE KEY（`clickTime`， `type`）

DISTRIBUTED BY HASH（`type`） BUCKETS 1

PROPERTIES （

“replication_allocation” = “tag.location.default： 1”

）;

創建 Routine Load 任務

CREATE ROUTINE LOAD pulsar_doris.load_from_pulsar_test ON clicklog

COLUMNS（clickTime，id，type，user）

PROPERTIES

（

“desired_concurrent_number”=“3”，

“max_batch_interval” = “20”，

“max_batch_rows” = “300000”，

“max_batch_size” = “209715200”，

“strict_mode” = “false”，

“format” = “json”

）

FROM KAFKA

（

“kafka_broker_list” = “127.0.0.1:9092”，

“kafka_topic” = “test”，

“property.group.id” = “doris”

）;

上述命令中的參數解釋如下：

pulsar_doris ：Routine Load 任務所在的數據庫

load_from_pulsar_test：Routine Load 任務名稱

clicklog：Routine Load 任務的目標表，也就是配置 Routine Load 任務將數據導入到 Doris 哪個表中。

strict_mode：導入是否為嚴格模式，這里設置為 False。

format：導入數據的類型，這里配置為 Json。

kafka_broker_list：Kafka Broker 服務的地址

kafka_broker_list：Kafka Topic 名稱，也就是同步哪個 Topic 上的數據。

property.group.id：消費組 ID

數據導入和測試

1. 數據導入 構造一個 ClickLog 的數據結構，并調用 Kafka 的 Producer 發送 5000 萬條數據到 Pulsar。 ClickLog 數據結構如下：

public class ClickLog {

private String id;

private String user;

private String city;

private String clickTime;

private String type;

。.. //省略getter和setter

}

消息構造和發送的核心代碼邏輯如下：

String strDateFormat = “yyyy-MM-dd HHss”;

@Autowired

private Producer producer;

try {

for（int j =0 ; j《50000;j++）{

int batchSize = 1000;

for（int i = 0 ; i《batchSize ;i++）{

ClickLog clickLog = new ClickLog（）;

clickLog.setId（UUID.randomUUID（）.toString（））;

SimpleDateFormat simpleDateFormat = new SimpleDateFormat（strDateFormat）;

clickLog.setClickTime（simpleDateFormat.format（new Date（）））;

clickLog.setType（“webset”）;

clickLog.setUser（“user”+ new Random（）.nextInt（1000） +i）;

producer.sendMessage（Constant.topicName， JSONObject.toJSONString（clickLog））;

}

}

} catch （Exception e） {

e.printStackTrace（）;

}

2. ROUTINE LOAD 任務查看執行 SHOW ALL ROUTINE LOAD FOR load_from_pulsar_test G;命令，查看導入任務的狀態。

mysql》 SHOW ALL ROUTINE LOAD FOR load_from_pulsar_test G;

*************************** 1. row ***************************

Id： 87873

Name： load_from_pulsar_test

CreateTime： 2022-05-31 1234

PauseTime： NULL

EndTime： NULL

DbName： default_cluster:pulsar_doris

TableName： clicklog1

State： RUNNING

DataSourceType： KAFKA

CurrentTaskNum： 1

JobProperties： {“partitions”：“*”，“columnToColumnExpr”：“clickTime，id，type，user”，“maxBatchIntervalS”：“20”，“whereExpr”：“*”，“dataFormat”：“json”，“timezone”：“Europe/London”，“send_batch_parallelism”：“1”，“precedingFilter”：“*”，“mergeType”：“APPEND”，“format”：“json”，“json_root”：“”，“maxBatchSizeBytes”：“209715200”，“exec_mem_limit”：“2147483648”，“strict_mode”：“false”，“jsonpaths”：“”，“deleteCondition”：“*”，“desireTaskConcurrentNum”：“3”，“maxErrorNum”：“0”，“strip_outer_array”：“false”，“currentTaskConcurrentNum”：“1”，“execMemLimit”：“2147483648”，“num_as_string”：“false”，“fuzzy_parse”：“false”，“maxBatchRows”：“300000”}

DataSourceProperties： {“topic”：“test”，“currentKafkaPartitions”：“0”，“brokerList”：“127.0.0.1:9092”}

CustomProperties： {“group.id”：“doris”，“kafka_default_offsets”：“OFFSET_END”，“client.id”：“doris.client”}

Statistic： {“receivedBytes”：5739001913，“runningTxns”：［］，“errorRows”：0，“committedTaskNum”：168，“loadedRows”：50000000，“loadRowsRate”：23000，“abortedTaskNum”：1，“errorRowsAfterResumed”：0，“totalRows”：50000000，“unselectedRows”：0，“receivedBytesRate”：2675000，“taskExecuteTimeMs”：2144799}

Progress： {“0”：“51139566”}

Lag： {“0”：0}

ReasonOfStateChanged：

ErrorLogUrls：

OtherMsg：

1 row in set （0.00 sec）

ERROR：

No query specified

從上面結果可以看到 totalRows 為 50000000，errorRows 為 0。說明數據不丟不重的導入 Apache Doris 了。

3. 數據統計驗證執行如下命令統計表中的數據，發現統計的結果也是 50000000，符合預期。

mysql》 select count（*） from clicklog;

+----------+

| count（*） |

+----------+

| 50000000 |

+----------+

1 row in set （3.73 sec）

mysql》

通過 KoP 我們實現了將 Apache Pulsar 數據無縫接入 Apache Doris ，無需對 Routine Load 任務進行任何修改，并保障了數據導入過程中的事務性。與此同時，Apache Doris 社區已經啟動了 Apache Pulsar 原生導入支持的設計，相信在不久后就可以直接訂閱 Pulsar 中的消息數據，并保證數據導入過程中的 Exactly-Once 語義。

審核編輯：郭婷