技术干货｜如何将 Pulsar 数据快速且无缝接入 Apache Doris

--------------------------|     Apache Doris       ||     ---------------    ||     | Routine Load |   ||     ---------------    |--------------------------            |Kafka Protocol(librdkafka)------------v--------------|     ---------------    ||     |     KoP      |   ||     ---------------    ||       Apache 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.gzcd apache-pulsar-2.10.0

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

mvn clean install -DskipTests

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

[root@17a5da45700b apache-pulsar-2.10.0]# ls protocols/pulsar-protocol-handler-kafka-2.11.0-SNAPSHOT.nar

#kop适配的协议messagingProtocols=kafka#kop 的NAR文件路径protocolHandlerDirectory=./protocols#是否允许自动创建topicallowAutoTopicCreationType=partitioned

# 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 defaultkafkaAdvertisedListeners=PLAINTEXT://127.0.0.1:9092brokerEntryMetadataInterceptors=org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptorbrokerDeleteInactiveTopicsEnabled=false

java.lang.IllegalArgumentException: Broker has disabled transaction coordinator, please enable it before using transaction.

kafkaTransactionCoordinatorEnabled=truetransactionCoordinatorEnabled=true

#前台启动#bin/pulsar standalone#后台启动pulsar-daemon start standalone

#进入Dorismysql -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 1PROPERTIES ("replication_allocation" = "tag.location.default: 1");

CREATE ROUTINE LOAD pulsar_doris.load_from_pulsar_test ON clicklogCOLUMNS(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" );

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 HH:mm:ss";       @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();        }

mysql>  SHOW ALL ROUTINE LOAD FOR load_from_pulsar_test \G;*************************** 1. row ***************************                  Id: 87873                Name: load_from_pulsar_test          CreateTime: 2022-05-31 12:03:34           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

mysql> select count(*) from clicklog;+----------+| count(*) |+----------+| 50000000 |+----------+1 row in set (3.73 sec)mysql>