关于 Apache Pulsar 在 Apache InLong 接入数据

> 本文由 [简悦 SimpRead](http://ksria.com/simpread/) 转码， 原文地址 [segmentfault.com](https://segmentfault.com/a/1190000041461176?utm_source=tuicool&utm_medium=referral)

> 关于 Apache PulsarApache Pulsar 是 Apache 软件基金会顶级项目，是下一代云原生分布式消息流平台，集消息、存储、轻量化函数式计算为一体，采用计算与存储...

关于 Apache Pulsar
----------------

Apache Pulsar 是 Apache 软件基金会顶级项目，是下一代云原生分布式消息流平台，集消息、存储、轻量化函数式计算为一体，采用计算与存储分离架构设计，支持多租户、持久化存储、多机房跨区域数据复制，具有强一致性、高吞吐、低延时及高可扩展性等流数据存储特性。  
GitHub 地址：[http://github.com/apache/pulsar/](https://link.segmentfault.com/?enc=7rB5Xs1OMp0FeGYr6hLdww%3D%3D.X09ZbH6DcgB2MP3J1pOPdrj224IdrI%2BTdHpie6XhJkNV%2FXD5Uk%2BschwIuwiTne2s)

> 文章转自公众号：Apache InLong，原文地址：[https://mp.weixin.qq.com/s/Wg...](https://link.segmentfault.com/?enc=58HgfsanBw2B%2Buj2F08%2BtQ%3D%3D.FDPUNGJR8RCjmH2jVblo2Jnowk%2ByPjUSikU%2BmBmXE9I%2FHmm%2Bo3CMqNq2%2FRCo4KtmNwJzIO0z0uQSB2LltdmJVQ%3D%3D)

Apache InLong 增加了通过 Apache Pulsar 接入数据的能力，充分利用了 Pulsar 不同于其它 MQ 的技术优势，为金融、计费等数据质量要求更高的数据接入场景，提供完整的解决方案。在下面的内容中，我们将通过一个完整的示例介绍如何通过 Apache InLong 使用 Apache Pulsar 接入数据。

![](http://kmgy.top:9090/image/2022/9/11/1460000041461178_repeat_1662900978414__143286.jpeg)

Apache InLong（incubating）简介
---------------------------

Apache InLong（应龙 [https://inlong.apache.org](https://link.segmentfault.com/?enc=EsSeonC7v1tLCZ7xL7H1RA%3D%3D.Pdqql981AYDHlwSKhe9FNFqe3dOgf%2Bgy1fm8f3R4qqc%3D)）是腾讯捐献给 Apache 社区的一站式数据流接入服务平台，提供自动、安全、可靠和高性能的数据传输能力，方便业务构建基于流式的数据分析、建模和应用。InLong 项目原名 TubeMQ ，专注于高性能、低成本的消息队列服务。为了进一步释放 TubeMQ 周边的生态能力，我们将项目升级为 InLong，专注打造一站式数据流接入服务平台。Apache InLong 以腾讯内部使用的 TDBank 为原型，依托万亿级别的数据接入和处理能力，整合了数据采集、汇聚、存储、分拣数据处理全流程，拥有简单易用、灵活扩展、稳定可靠等特性。

![](http://kmgy.top:9090/image/2022/9/11/1460000041461179_repeat_1662900978185__621869.jpeg)

Apache InLong 服务于数据采集到落地的整个生命周期，按数据的不同阶段提供不同的处理模块，主要包括：

*   inlong-agent，数据采集 Agent，支持从指定目录或文件读取常规日志、逐条上报。后续也将扩展 DB 采集、HTTP 上报等能力；
*   inlong-dataproxy，一个基于 Flume-ng 的 Proxy 组件，支持数据发送阻塞和落盘重发，拥有将接收到的数据转发到不同 MQ（消息队列）的能力；
*   inlong-tubemq，腾讯自研的消息队列服务，专注于大数据场景下海量数据的高性能存储和传输，在海量实践和低成本方面有着良好的核心优势；
*   inlong-sort，对从不同的 MQ 消费到的数据进行 ETL 处理，然后汇聚并写入 Hive、ClickHouse、Hbase、Iceberg 等存储系统；
*   inlong-manager，提供完整的数据服务管控能力，包括元数据、任务流、权限，OpenAPI 等；
*   inlong-website，用于管理数据接入的前端页面，简化整个 InLong 管控平台的使用。

关于 Apache Pulsar
----------------

![](http://kmgy.top:9090/image/2022/9/11/1460000041461180_repeat_1662900978183__474774.jpeg)

Apache Pulsar 是 Pub/Sub 模型的消息系统，并且从设计上做了存储和计算的分离。Apache Pulsar 计算与存储分离的架构，以及分片存储的设计为 Apache Pulsar 带来了相比于传统基于分区存储 MQ 的一些优势:

*   Broker 和 Bookie 相互独立，方便实现独立的扩展以及独立的容错；
*   Broker 无状态，便于快速上、下线，更加适合于云原生场景；
*   分区存储不受限于单个节点存储容量；
*   分区数据分布均匀。

准备条件
----

*   安装 Apache Pulsar，版本 2.6+
*   安装 Apache Hive，版本 2.3+

安装 InLong
---------

部署 InLong ，可以使用 Docker Compose 实现一键部署，也可以通过二进制文件在普通机器上部署。

*   Docker Compose 部署：[https://inlong.apache.org/zh-...](https://link.segmentfault.com/?enc=hg59P3n9uI4EHNiwya%2BOLQ%3D%3D.LlPnWL6hnLZjID6Q2WZT01GX0%2BQxM49rpGRKtykxN9vjXO2jm8XWEs5uK7rRZRqQtYKthOSsXeeA8PugHRBnAg%3D%3D)
*   使用安装包部署：[https://inlong.apache.org/zh-...](https://link.segmentfault.com/?enc=ZBq5dK9ox9grU5fdMhCqXw%3D%3D.8a7wTNcnfZZd1vAKPCyC%2BiEAd5uoKb9GjT0%2FihIqEhT%2BOQL87tte3n6fE6%2FgomlMd98MP0RtSRTHWCJMzLPfFA%3D%3D)

区别于 InLong TubeMQ，如果使用 Apache Pulsar，需要在 Manager 组件安装中配置 Pulsar 集群信息，格式如下：

```
# Pulsar admin URL
pulsar.adminUrl=http://127.0.0.1:8080,127.0.0.2:8080,127.0.0.3:8080
# Pulsar broker address
pulsar.serviceUrl=pulsar://127.0.0.1:6650,127.0.0.1:6650,127.0.0.1:6650
# Default tenant of Pulsar
pulsar.defaultTenant=public
```

创建数据接入
------

#### 配置数据流 Group 信息

![](http://kmgy.top:9090/image/2022/9/11/1460000041461181_repeat_1662900978183__138457.jpeg)

在创建数据接入时，数据流 Group 可选用的消息中间件选择 Pulsar，其它跟 Pulsar 相关的配置项还包括：

*   Queue module：队列模型，并行或者顺序，选择并行时可设置 Topic 的分区数，顺序则为一个分区；
*   Write quorum：消息写入的副本数；
*   Ack quorum：确认写入 Bookies 的数量；
*   retention time：已被 consumer 确认的消息被保存的时间；
*   ttl：未被确认的消息的过期时间；
*   retention size：已被 consumer 确认的消息被保存的大小。

#### 配置数据流

![](http://kmgy.top:9090/image/2022/9/11/1460000041461182_repeat_1662900978180__770056.jpeg)

配置消息来源时，文件数据源中的文件路径，可参照 inlong-agent 中 [File Agent 的详细指引](https://link.segmentfault.com/?enc=NFI2od6nDTwNwP4n%2BTPAdQ%3D%3D.OQwvyWBUNeo0fFwcwx4QG2aT7JdoeZxcvMHwc%2BnqjG4N7SVOSYoI59HekcM6K7DInAi50UfEMRjje4ygcRnvKsDazM7q0ClV6nmSiV%2FjArk%3D)。

#### 配置数据格式

![](http://kmgy.top:9090/image/2022/9/11/1460000041461183_repeat_1662900978436__427959.jpeg)

#### 配置 Hive 集群

保存 Hive 流向，点击 “提交审批”。

![](http://kmgy.top:9090/image/2022/9/11/1460000041461184_repeat_1662900978731__949373.jpeg)

数据接入审批
------

进入审批管理页面，点击我的审批，审批上面提交的接入申请，审批结束后会在 Pulsar 集群同步创建数据流需要的 Topic 和订阅。

我们可以在 Pulsar 集群使用命令行工具检查 Topic 是否创建成功：

![](http://kmgy.top:9090/image/2022/9/11/1460000041461185_repeat_1662900979043__853321.jpeg)

配置文件 Agent
----------

在配置文件 Agent 时，需要根据数据接入创建时指定的目录下创建文件：

`touch /data/test_file.txt;`

按照创建数据流时的数据源格式，向文件中写入数据（可以按格式写入更多数据）：

`echo -e "1|test\n2|test\n" >> /data/test_file.txt`

数据落地检查
------

最后，我们登入 Hive 集群，通过 Hive 的 SQL 命令查看 `test_stream` 表中是否成功插入了数据。

问题排查
----

如果出现数据未正确写入 Hive 集群，可以检查 Dataproxy 和 Sort 相关信息是否同步：

*   检查 Inlong-Dataproxy 的 conf/topics.properties 文件夹中是否正确写入该数据流对应的 Topic 信息`b_test_group/test_stream=persistent://public/b_test_group/test_stream`
*   检查 InLong Sort 监听的 ZooKeeper 中是否成功推送了数据流的配置信息：`get /inlong_hive/dataflows/{{sink_id}}`