Pulsar#

简介#

Apache Pulsar是一个云原生的,分布式消息流平台。它可以作为OpenMLDB的在线数据源,将实时的数据流导入到OpenMLDB在线。了解更多Pulsar,请参考官网https://pulsar.apache.org/。我们开发了在Pulsar中使用的OpenMLDB JDBC Connector,可以无障碍地使Pulsar连接到OpenMLDB。在这篇文档中,你将学习到这个connector的概念与使用方法。

注意,为了使演示更简单,本文中将使用Pulsar Standalone,OpenMLDB集群和一个简单JSON消息生产者程序,来演示OpenMLDB JDBC Connector是如何工作的。该connector是完全可以在Pulsar Cluster中正常使用的。

See also

关于 Pulsar 的 OpenMLDB Connector 的详细信息,也可以参考 Pulsar 官网相关介绍

概览#

下载#

  • 你需要下载本文中所需要的所有文件,请点击files下载。文件包括connector包,schema文件,配置文件等等。

  • 如果你只想要下载connector包用于自己的项目,请点击connector snapshot

流程#

使用connector的简要流程,如下图所示。我们接下来将详细介绍每一步。我们也录制了所有步骤,详情见terminalizer分享, 你也可以在此下载录制的脚本demo.yml

整体上,使用流程可以概括为三步:

  1. 在 OpenMLDB 创建相关的数据库和表

  2. 在 Pulsar 创建 sink,来把 Pulsar 数据流和 OpenMLDB 连接起来,同时需要在 Pulsar 配置相应的 schema,使得数据流可以正确的被 OpenMLDB 接收并且存入到线上数据库。

  3. 进行测试或者正常使用

demo steps

步骤 1:在 OpenMLDB 创建数据库和数据表#

启动 OpenMLDB 集群#

使用Docker可以快速启动OpenMLDB,除此之外,我们还需要创建测试用的表。创建 OpenMLDB 详请可以参考集群版 OpenMLDB 快速上手

Caution

目前只有OpenMLDB集群版可以作为sink的接收端,数据只会sink到集群的在线存储中。

我们更推荐你使用‘host network’模式运行docker,以及绑定文件目录‘files’,sql脚本在该目录中。

docker run -dit --network host -v `pwd`/files:/work/pulsar_files --name openmldb 4pdosc/openmldb:0.8.5 bash
docker exec -it openmldb bash

在OpenMLDB容器中,启动集群:

./init.sh

Caution

在macOS平台上,即使使用host网络,也不支持从容器外部去连接容器内的 OpenMLDB 服务器。但从容器内,去连接别的容器内的OpenMLDB服务,是可行的。

创建表#

我们使用一个脚本快速创建表,脚本内容如下:

create database pulsar_test;
use pulsar_test;
create table connector_test(id string, vendor_id int, pickup_datetime bigint, dropoff_datetime bigint, passenger_count int, pickup_longitude double, pickup_latitude double, dropoff_longitude double, dropoff_latitude double, store_and_fwd_flag string, trip_duration int);
desc connector_test;

执行脚本:

/work/openmldb/bin/openmldb --zk_cluster=127.0.0.1:2181 --zk_root_path=/openmldb --role=sql_client < /work/pulsar_files/create.sql

table desc

Note

目前,Pulsar中JSONSchema和JDBC base connector都不支持’java.sql.Timestamp’。所以我们使用’long’作为timestamp列的数据类型(在OpenMLDB可以使用long作为时间戳)。

步骤 2:在 Pulsar 创建 sink 和 schema#

启动 Pulsar Standalone#

使用docker,可以更简单快速的启动Pulsar。我们推荐你使用’host network’来运行docker,这样可以避免诸多容器相关的网络连接问题。而且,我们需要使用pulsar-admin来进行sink创建,这个程序在Pulsar镜像内。所以,我们使用bash运行容器,在容器内部逐一执行命令。此处,也需要绑定’files’文件目录。

docker run -dit --network host -v `pwd`/files:/pulsar/files --name pulsar apachepulsar/pulsar:2.9.1 bash
docker exec -it pulsar bash

在Pulsar容器中,启动standalone服务端。

bin/pulsar-daemon start standalone --zookeeper-port 5181

Note

OpenMLDB服务已经使用了端口2181,所以此处我们为Pulsar重新设置一个zk端口。我们将使用端口2181来连接OpenMLDB,但Pulsar standalone内的zk端口不会对外造成影响。

你可以使用ps检查Pulsar是否正常运行,如果启动失败,检查日志logs/pulsar-standalone-....log

ps axu|grep pulsar

当你启动一个本地standalone集群,会自动创建’pulic/default’ namesapce。这个namespace用于开发,参考pulsar文档.

我们将在此namespace中创建sink

See also

如果你想要在本地直接启动Pulsar,参考Set up a standalone Pulsar locally.

Q&A#

Q: 碰到以下问题是什么原因

2022-04-07T03:15:59,289+0000 [main] INFO  org.apache.zookeeper.server.NIOServerCnxnFactory - binding to port 0.0.0.0/0.0.0.0:5181
2022-04-07T03:15:59,289+0000 [main] ERROR org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble - Exception while instantiating ZooKeeper
java.net.BindException: Address already in use

A: Pulsar需要一个未被使用的端口来启动zk,端口5181页已经被使用,需要再更改一下’–zookeeper-port’的端口号。

Q: 8080端口已被使用?

A: 8080是’webServicePort’默认配置端口,在conf/standalone.conf中,可以更换这个端口。但注意,pulsar-admin会使用conf/client.conf中的’webServiceUrl’进行连接,也需要同步更改。

Q: 6650端口已被使用?

A: 需要同步更改conf/standalone.conf中的’brokerServicePort’和conf/client.conf中的’brokerServiceUrl’配置项。

Connector安装(Optional)#

前面的步骤中我们绑定了’files’目录,里面已经提供了connector的nar包。我们可以使用“非内建connector”模式来设置connector(即在sink配置中指定’archive’配置项,将在下一个步骤中描述)。

但如果你希望将OpenMLDB connector作为内建的connector,你需要创建’connectors’目录,并拷贝nar文件到’connectors’目录。

mkdir connectors
cp files/pulsar-io-jdbc-openmldb-2.11.0-SNAPSHOT.nar  connectors/

如果在Pulsar运行时,你想改变或增加connector,你可以通知Pulsar更新信息:

bin/pulsar-admin sinks reload

当OpenMLDB connector成为内建connector时,它的sink类型名为’jdbc-openmldb’,你可以直接使用这个类型名来指定使用OpenMLDB connector。

创建sink#

我们使用’public/default’这个namespace来创建sink, 我们需要一个sink的配置文件, 它在files/pulsar-openmldb-jdbc-sink.yaml,内容如下:

 tenant: "public"
 namespace: "default"
 name: "openmldb-test-sink"
 archive: "files/pulsar-io-jdbc-openmldb-2.11.0-SNAPSHOT.nar"
 inputs: ["test_openmldb"]
 configs:
     jdbcUrl: "jdbc:openmldb:///pulsar_test?zk=localhost:2181&zkPath=/openmldb"
     tableName: "connector_test"

Note

‘name’:sink名。

‘archive’:我们使用’archive’来指定sink connector, 所以这里我们是将OpenMLDB connector当作非内建connector使用。

‘input’:可以是多个topic的名字,本文只使用一个。

‘config’:用于连接OpenMLDB集群的jdbc配置。

接下来,创建这个sink并检查。注意,我们设置的输入topic是’test_openmldb’,后续步骤需要使用到。

./bin/pulsar-admin sinks create --sink-config-file files/pulsar-openmldb-jdbc-sink.yaml
./bin/pulsar-admin sinks status --name openmldb-test-sink

init sink status

创建 Schema#

上传schema到topic ‘test_openmldb’,schema类型是JSON格式。后续步骤中,我们将生产一样schema的JSON消息。schema文件是files/openmldb-table-schema,内容如下:

{
    "type": "JSON",
    "schema":"{\"type\":\"record\",\"name\":\"OpenMLDBSchema\",\"namespace\":\"com.foo\",\"fields\":[{\"name\":\"id\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"vendor_id\",\"type\":\"int\"},{\"name\":\"pickup_datetime\",\"type\":\"long\"},{\"name\":\"dropoff_datetime\",\"type\":\"long\"},{\"name\":\"passenger_count\",\"type\":\"int\"},{\"name\":\"pickup_longitude\",\"type\":\"double\"},{\"name\":\"pickup_latitude\",\"type\":\"double\"},{\"name\":\"dropoff_longitude\",\"type\":\"double\"},{\"name\":\"dropoff_latitude\",\"type\":\"double\"},{\"name\":\"store_and_fwd_flag\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"trip_duration\",\"type\":\"int\"}]}",
    "properties": {}
}

上传并检查schema的命令,如下所示:

./bin/pulsar-admin schemas upload test_openmldb -f ./files/openmldb-table-schema
./bin/pulsar-admin schemas get test_openmldb

topic schema

步骤 3:测试#

发送消息#

我们使用两条OpenMLDB镜像中data/taxi_tour_table_train_simple.csv的样本数据,作为测试用的消息。数据如下图所示: test data

Java Producer#

Producer JAVA代码,详情见demo producer。核心代码如下: snippet

可以看到,producer将发送两条消息到topic ‘test_openmldb’。这之后,Pulsar将读到消息,并将其写入OpenMLDB集群的在线存储中。

程序包在’files’中,你可以直接运行它:

java -cp files/pulsar-client-java-1.0-SNAPSHOT-jar-with-dependencies.jar org.example.Client

Python Producer#

Producer也可以使用Python实现,详情见files/pulsar_client.py。运行前需要安装pulsar python client:

pip3 install pulsar-client==2.9.1

运行:

python3 files/pulsar_client.py

Note

已知问题:消息中如果Long类型的value较小,在pulsar function sink的jdbc bindValue阶段,会从JsonRecord中读出java Integer Type的值,从而setColumnValue时对Long类型的列做statement.setInt而不是setLong。因此,产生data type not match的SQLException。 如果想要查看具体哪一列的data type有问题,请打开function的debug level日志,方法见调试

检查#

在Pulsar中检查#

我们可以检查Pulsar中的sink状态:

./bin/pulsar-admin sinks status --name openmldb-test-sink 

sink status

Note

“numReadFromPulsar”: pulsar发送了2条message到sink实例中。

“numWrittenToSink”: sink实例向OpenMLDB写入2条message。

在OpenMLDB中检查#

我们可以在OpenMLDB在线存储中查询到这些消息数据。查询脚本select.sql内容如下:

set @@execute_mode='online';
use pulsar_test;
select *, string(timestamp(pickup_datetime)), string(timestamp(dropoff_datetime)) from connector_test;

在OpenMLDB容器中执行脚本:

/work/openmldb/bin/openmldb --zk_cluster=127.0.0.1:2181 --zk_root_path=/openmldb --role=sql_client < /work/pulsar_files/select.sql

openmldb result

调试#

如果OpenMLDB没有数据,而sinks status合理,那么sink写入应该出了问题。请查看sink日志,日志地址为logs/functions/public/default/openmldb-test-sink/openmldb-test-sink-0.log。如果你使用别的sink name,请定位到正确的sink日志。

Pulsar会重复尝试写入此前没有写入成功的消息,所以,如果你曾发送了错误的消息,即使新消息成功写入,日志中仍然会有错误消息的写入报错。测试中推荐直接truncate topic再测试:

./bin/pulsar-admin topics truncate persistent://public/default/test_openmldb

如果自命名的sink name,可通过./bin/pulsar-admin topics list public/default查询topic全名。

debug日志#

如果sink日志信息不足够定位,可以打开debug日志。需要修改配置,并重启sink。vim conf/functions_log4j2.xml并做一下修改:

        <Property>
            <name>pulsar.log.level</name>
            <value>debug</value> <!-- 设置为debug level -->
        </Property>
        <Root>
            <level>${sys:pulsar.log.level}</level> <!-- 此处info改为${sys:pulsar.log.level}或debug -->
            <AppenderRef>
                <ref>${sys:pulsar.log.appender}</ref>
                <level>${sys:pulsar.log.level}</level>
            </AppenderRef>
        </Root>

再重新启动sink:

./bin/pulsar-admin sinks restart --name openmldb-test-sink

重建pulsar#

bin/pulsar-daemon stop standalone --zookeeper-port 5181
rm -r data logs
bin/pulsar-daemon start standalone --zookeeper-port 5181