如何使用 Apache Flume 发送日志数据至 Apache Pulsar

阅读本文需要约 7 分钟。

Apache Flume 是一个分布式的、可靠易用的系统，可以有效地收集和汇总来自多种源系统的大量日志数据，或转移这些数据至一个数据中心存储。

Apache Pulsar 是 Yahoo 基于 Apache BookKeeper 开发和开源的下一代分布式消息系统。Apache Pulsar 已经从下一代分布式消息系统演化成为一个流原生数据平台。

本文主要介绍使用 Flume 实现日志搜集，并发送日志数据至 Pulsar 进行消费。整体架构如下： 

本文主要涉及 Flume 和 Pulsar 服务的搭建、测试、数据的发送及消费，全面讲解了 Flume 如何发送数据到 Pulsar以及Pulsar 如何消费数据。上图虽然画了 3 个 Flume 服务，但是本次测试只使用了一个，目的是快速走通整个流程，消费数据时也是只启动了一个消费端。后期如果有需求，可以根据此次试验快速完成扩展。

安装环境依赖

在搭建本次试验之前，需要安装以下依赖，本次试验是在 Mac 系统上进行的测试。

• Docker

  https://www.docker.com/get-started  

• Java 8  https://www.oracle.com/technetwork/java/javase/downloads/index.html  

• Maven 3.5 或更高版本  https://archive.apache.org/dist/maven/maven-3/ 

• Git  https://www.linode.com/docs/development/version-control/how-to-install-git-on-linux-mac-and-windows/    

• Telnet    

  http://www.telnet.org/htm/howto.htm  

编译并打包项目

你能通过以下任一方式编译并打包项目：

方法 1：在本地编译并打包本项目

注意

在打包本项目之前，需要安装并配置好 Java 、 Maven 和 Git 的相关环境。

1、从 GitHub 克隆 Flume sink 代码。

git clone https://github.com/streamnative/pulsar-flume-ng-sink.git

2、使用 Maven 将本项目打包。

1  cd pulsar-flume-ng-sink
2  mvn clean package

如果出现以下信息，则说明打包成功：

1  [INFO] ------------------------------------------------------------------------
2  [INFO] BUILD SUCCESS
3  [INFO] ------------------------------------------------------------------------

方法 2：使用 Docker 镜像编译

并打包本项目

如果本地不方便安装 Java 和 Maven 环境，可以使用 Docker 环境进行打包。

1、拉取并启动镜像。

本次使用的 Docker 镜像是 maven:3.6-jdk-8 ，带有 Maven 、 Java 和 Git 的相关环境。

1  docker pull maven:3.6-jdk-8
2  docker run -d -it --name docker-package maven:3.6-jdk-8 /bin/bash

上述命令会启动一个基于镜像 maven:3.6-jdk-8 名称为 docker-package 的 Docker 服务。

1  docker ps -a | grep docker-package
2  580376b38aa8        maven:3.6-jdk-8                              "/usr/local/bin/mvn-…"   5 minutes ago       Up 5 minutes                                    docker-package
3  docker exec -it docker-package /bin/bash

使用上述命令确认该镜像是否成功启动。成功启动后，我们就已经在 Docker 服务中了。

2、克隆代码并打包。

成功启动该镜像后，可以非常方便地进行打包。类似地，我们可以使用与在本地打包相同的方法，在 Docker 中对本项目进行打包。

1  git clone https://github.com/streamnative/pulsar-flume-ng-sink.git
2  cd pulsar-flume-ng-sink/
3  mvn clean package

如果出现以下信息，则说明打包成功：

1  [INFO] ------------------------------------------------------------------------
2  [INFO] BUILD SUCCESS
3  [INFO] ------------------------------------------------------------------------

打包成功后，在本目录下的 target 文件夹中会出现一个 jar 包，名字类似于  flume-ng-pulsar-sink-<version>.jar。

安装 Pulsar

为了方便测试，本次安装 standalone Pulsar，并且使用 Docker 镜像 apachepulsar/pulsar:2.3.0 启动服务。

1docker pull apachepulsar/pulsar:2.3.0
2docker run -d -it -p 6650:6650 -p 8080:8080 -v $PWD/data:/pulsar/data --name pulsar-flume-standalone apachepulsar/pulsar:2.3.0 bin/pulsar standalone

该服务暴露 6650 和 8080 端口供外部服务使用，该容器将 Pulsar 服务的 data 目录挂载在了当前目录下，并且命名为 pulsar-flume-standalone。 -d 参数表示使该服务使用后台模式运行；-it 参数表示以交互模式运行容器，并为容器分配一个伪输入终端。

1docker logs -f pulsar-flume-standalone
2
307:53:43.844 [pulsar-web-55-6] INFO  org.apache.pulsar.broker.web.PulsarWebResource - Successfully validated clusters on tenant [public]
407:53:43.856 [pulsar-web-55-6] INFO  org.apache.pulsar.broker.admin.impl.NamespacesBase - [null] Created namespace public/default
507:53:43.859 [pulsar-web-55-6] INFO  org.eclipse.jetty.server.RequestLog - 172.17.0.3 - - [28/Apr/2019:07:53:43 +0000] "PUT /admin/v2/namespaces/public/default HTTP/1.1" 204 0 "-" "Jersey/2.27 (HttpUrlConnection 1.8.0_181)" 17
607:53:43.864 [pulsar-web-55-6] INFO  org.apache.pulsar.broker.web.PulsarWebResource - Successfully validated clusters on tenant [public]
707:53:43.868 [pulsar-ordered-OrderedExecutor-1-0-EventThread] INFO  org.apache.pulsar.zookeeper.ZooKeeperDataCache - [State:CONNECTED Timeout:30000 sessionid:0x1000112d2ea000a local:/127.0.0.1:38464 remoteserver:localhost/127.0.0.1:2181 lastZxid:154 xid:41 sent:41 recv:43 queuedpkts:0 pendingresp:0 queuedevents:0] Received ZooKeeper watch event: WatchedEvent state:SyncConnected type:NodeDataChanged path:/admin/policies/public/default
807:53:43.869 [pulsar-web-55-6] INFO  org.apache.pulsar.broker.admin.impl.NamespacesBase - [null] Successfully updated the replication clusters on namespace public/default

使用 docker logs -f pulsar-flume-standalone 命令查看 Docker 日志。如果出现以上的日志信息，则说明 Pulsar 在单机模式下启动成功了。

使用 Flume

使用 Docker 搭建 Flume 环境。

1、启动 Flume 容器

1 docker pull maven:3.6-jdk-8
2 docker run -d -it --link pulsar-flume-standalone -p 44445:44445 --name flume maven:3.6-jdk-8 /bin/bash

以上命令会启动一个名称为 flume 的服务，它链接到 pulsar-flume-standalone 容器，暴露 44445 端口的 Flume 服务。

1  docker ps -a | grep flume
2  27c7555e5481        maven:3.6-jdk-8                              "/usr/local/bin/mvn-…"   46 seconds ago      Up 45 seconds               0.0.0.0:44445->44445/tcp                         flume

使用 docker ps 命令可以看到启动的 Flume 服务。

2、安装 Flume 服务

（1）进入 Flume 容器。

docker exec -it flume /bin/bash

（2）下载并解压 Apache Flume 压缩包到当前目录。

1  wget http://apache.01link.hk/flume/1.9.0/apache-flume-1.9.0-bin.tar.gz
2  tar -zxvf apache-flume-1.9.0-bin.tar.gz

（3）退出 Flume 容器。

3、部署 pulsar-flume-ng-sink

（1）修改配置文件。本次测试需使用以下 2 个配置文件：

配置文件 flume-example.conf （https://github.com/streamnative/pulsar-flume-ng-sink/blob/master/src/test/resources/flume-example.conf）

flume-example.conf 中兼容 Flume 的配置。更多关于 Flume 配置的信息，参阅 Flume 官方文档，此处简单介绍以下信息：

 1    # Describe/configure the source
 2    a1.sources.r1.type = netcat
 3    a1.sources.r1.bind = 0.0.0.0
 4    a1.sources.r1.port = 44445
 5
 6    ## Describe the sink
 7    a1.sinks.k1.type = org.apache.flume.sink.pulsar.PulsarSink
 8    # Configure the pulsar service url (without `pulsar://`)
 9    a1.sinks.k1.serviceUrl = pulsar-flume-standalone:6650
10    a1.sinks.k1.topicName = flume-test-topic
11    a1.sinks.k1.producerName = flume-test-producer

参数说明：

• a1.sources.r1.type    

  指定数据来源。为了方便测试，此处指定 netcat 可以使用 Telnet 工具进行测试。   

• a1.sources.r1.bind    

  绑定本机的 IP 地址。   

• a1.sources.r1.port    

  绑定本机的 44445 端口。启动容器会映射该端口映射，方便测试。   

• a1.sinks.k1.type    

  指定使用的 PulsarSink 类。这个不需要更改。   

• a1.sinks.k1.serviceUrl    

  指定 Pulsar 服务的地址。你可以根据实际需求，更改该参数。此处使用容器地址是 pulsar-flume-standalone:6650。如需支持 SSL，可以配置打开 useTLS 选项。   

• a1.sinks.k1.topicName    

  指定需要发送至 Pulsar 的目标 Topic。如果 Pulsar 服务没有该 Topic ，则会自动新建该 Topic。   

• a1.sinks.k1.producerName    

  指定 sink 的名称，不能重复。 

以上是一些比较简单的配置，根据这些配置，我们可以完成本次试验。更多关于生产环境参数的置、客户端配置和 Producer 配置等信息，参阅 pulsar-flume-ng-sink 的配置 （https://github.com/streamnative/pulsar-flume-ng-sink#configurations）。

配置文件 flume-env.sh （https://github.com/streamnative/pulsar-flume-ng-sink/blob/master/src/test/resources/flume-env.sh）

flume-env.sh 文件是 Flume 在启动的时需要使用的一些库的查找路径，需要将已打包好的 flume-ng-pulsar-sink-<version>.jar 的 jar 包复制到该路径下。

以下是本次测使用的 flume-env.sh 配置：

1      #!/bin/sh
2      export JAVA_HOME=/docker-java-home
3
4      FLUME_CLASSPATH=/docker-java-home/lib

（2）复制配置文件至 Flume 服务器。

按照以上步骤修改完配置文件后，可以将这些配置复制到 Flume 服务所在的容器。

1  docker cp flume-example.conf flume:/apache-flume-1.9.0-bin/conf/
2  docker cp flume-env.sh flume:/apache-flume-1.9.0-bin/conf/

（3）复制 flume-ng-pulsar-sink-.jar 至 Flume 容器。

使用以下任一方法：

方法1

如果在本地进行了打包，使用以下命令可以将该 jar 包复制到 Flume 容器。进入打包文件的根目录，使用 Docker 命令进行复制。

1    cd pulsar-flume-ng-sink
2    docker cp target/flume-ng-pulsar-sink-1.9.0.jar flume:apache-flume-1.9.0-bin/lib/

方法 2

如果在容器中进行了打包，首先将容器中已打包的 jar 包复制到本地，再使用 Docker 命令复制到 Flume 容器中。

1    docker cp docker-package:/pulsar-flume-ng-sink/target/flume-ng-pulsar-sink-1.9.0.jar .
2    docker cp flume-ng-pulsar-sink-1.9.0.jar flume:apache-flume-1.9.0-bin/lib/

至此，已成功构建启动 Flume 服务所需要的环境和配置文件。

4、启动 Flume 服务

进入 Flume 容器并启动 flume agent。

1  docker exec -it flume /bin/bash
2  apache-flume-1.9.0-bin/bin/flume-ng agent --conf apache-flume-1.9.0-bin/conf/ -f apache-flume-1.9.0-bin/conf/flume-example.conf -n a1

在以上命令中，--conf 指定 Flume 服务的配置文件夹，-f 指定配置文件，-n 指定名称（需要与配置文件里的相同）。

如果出现如下信息，则说明 flume-ng agent 启动成功。

1  Info: Sourcing environment configuration script /apache-flume-1.9.0-bin/conf/flume-env.sh
2  Info: Including Hive libraries found via () for Hive access
3  + exec /docker-java-home/bin/java -Xmx20m -cp '/apache-flume-1.9.0-bin/conf:/apache-flume-1.9.0-bin/lib/*:/docker-java-home/lib:/lib/*' -Djava.library.path= org.apache.flume.node.Application -f apache-flume-1.9.0-bin/conf/flume-example.conf -n a1

5、启动消费端

在新的窗口中进入 Pulsar 的容器，再启动消费端（用于消费 Flume 发出的信息）。

测试使用的脚本文件在这里（https://github.com/streamnative/pulsar-flume-ng-sink/blob/master/src/test/python/pulsar-flume.py）。脚本文件内容如下：

 1  import pulsar
 2
 3  client = pulsar.Client('pulsar://localhost:6650')
 4  consumer = client.subscribe('flume-test-topic',
 5                              subscription_name='sub')
 6
 7  while True:
 8      msg = consumer.receive()
 9      print("Received message: '%s'" % msg.data())
10      consumer.acknowledge(msg)
11
12  client.close()

首先，该脚本会连接到本地 Pulsar 服务的 6650 端口。由于在 Pulsar 服务所在的容器中启动的该脚本，因此，服务地址是 localhost:6650。该脚本会在 flume-test-topic Topic 上进行消费，该 Topic 正是我们在 flume-example.conf 中配置的 Topic ，并且指定订阅名称为 sub。

（1）复制测试脚本文件到 Pulsar 服务所在的容器。

1    cd pulsar-flume-ng-sink
2    docker cp src/test/python/pulsar-flume.py pulsar-flume-standalone:/pulsar/

（2）进入 Pulsar 服务容器。

    docker exec -it pulsar-flume-standalone /bin/bash

（3）启动消费者脚本。

1      python pulsar-flume.py
2      2019-04-28 09:01:43.581 INFO  Client:88 | Subscribing on Topic :flume-test-topic
3      2019-04-28 09:01:43.582 INFO  ConnectionPool:72 | Created connection for pulsar://localhost:6650
4      2019-04-28 09:01:43.585 INFO  ClientConnection:300 | [127.0.0.1:56088 -> 127.0.0.1:6650] Connected to broker
5      2019-04-28 09:01:43.590 INFO  HandlerBase:52 | [persistent://public/default/flume-test-topic, sub, 0] Getting connection from pool
6      2019-04-28 09:01:43.590 INFO  ConnectionPool:72 | Created connection for pulsar://c063d9580d4c:6650
7      2019-04-28 09:01:43.591 INFO  ClientConnection:302 | [127.0.0.1:56090 -> 127.0.0.1:6650] Connected to broker through proxy. Logical broker: pulsar://c063d9580d4c:6650
8      2019-04-28 09:01:43.614 INFO  ConsumerImpl:169 | [persistent://public/default/flume-test-topic, sub, 0] Created consumer on broker [127.0.0.1:56090 -> 127.0.0.1:6650]

启动脚本后出现以上信息，则说明消费端已经成功启动。

6、发送消息进行测试

如前文所述，本次测试使用 Telnet 工具。

开启一个新的窗口，使用以下 Telnet 命令，发送数据到 Flume 的 44445 端口。

1  ➜  ~ telnet localhost 44445
2  Trying ::1...
3  Connected to localhost.
4  Escape character is '^]'.
5  hello
6  OK
7  world
8  OK

此时，在刚才开启的消费端可以看到通过 Telnet 发送的信息，内容大致如下：

1  'eceived message: 'hello
2  'eceived message: 'world

以上正是通过 Telnet 发送到 Flume 44445 端口的数据。

清理环境

完成本次测试后，使用以下命令停止并且删除相关容器：

1docker stop flume
2docker rm flume
3docker stop pulsar-flume-standalone
4docker rm pulsar-flume-standalone
5docker stop docker-package
6docker rm docker-package

更多信息

Apache Pulsar Connector 系列文章：

作者 | tuteng

审校 | Anonymitaet

编辑 | Irene

更多关于 Pulsar 的技术干货和产品动态，请关注 ApachePulsar 微信公众号。