streamsets 集成 rabbitmq 以及benthos stream 处理框架

streamsets 集成 rabbitmq 以及benthos stream 处理框架

benthos 是一个stream 处理框架，streamsets 也是，但是两者可以通过不同的工具进行集成起来
一般我们可以使用http 服务，消息中间件（kafka、rabbitmq 。

。

。

）

使用docker-compose 运行

服务配置

• docker-compose 文件

version: "3"

services:

  sets:

    image: streamsets/datacollector

    volumes:

     - "./ms/data:/data"

     - "./ms/logs:/logs"

     - "./ms/tmp:/tmp"

    ports:

     - "8000:8000"

     - "18630:18630"

  redis:

    image: redis

    ports:

     - "6379:6379"

  benthos-in:

    image: jeffail/benthos

    environment:

      INPUT_TYPE: "http_server"

      INPUT_HTTP_SERVER_ADDRESS: "0.0.0.0:8080"

      OUTPUT_TYPE: "amqp"

      OUTPUT_AMQP_URL: "amqp://guest:guest@rabbitmq:5672/appdemo"

      OUTPUT_AMQP_EXCHANGE: "benthos-exchange"

      OUTPUT_AMQP_EXCHANGE_TYPE: "direct"

      OUTPUT_AMQP_KEY: "benthos-key"

    ports:

      - "8080:8080"

  rabbitmq:

    image: rabbitmq:3.6.14-management

    ports:

      - "4369:4369"

      - "5671:5671"

      - "5672:5672"

      - "25672:25672"

      - "15672:15672"

• 启动

docker-compose up -d

• 添加streamsets rabbitmq 包
  
• rabbitmq vhost 创建
  

说明

benthos 使用http service 与rabbitmq 集成的模式，streamsets 使用rabbitmq customer 与local fs 集成的模式
benthos 服务的启动 curl http://localhost:8080/post -d "example message"
同时首选需要进行vhost 的创建，目前发现的一个bug的使用默认的就会有问题，需要单独创建，可以使用管理界面
http://docker-host:15672

streamsets 配置

• pipeline flow
  
• rabbitmq 配置
  

• local fs
  
• benthos 配置说明

参考docker-compose 文件

environment:

      INPUT_TYPE: "http_server"

      INPUT_HTTP_SERVER_ADDRESS: "0.0.0.0:8080"

      OUTPUT_TYPE: "amqp"

      OUTPUT_AMQP_URL: "amqp://guest:guest@rabbitmq:5672/appdemo"

      OUTPUT_AMQP_EXCHANGE: "benthos-exchange"

      OUTPUT_AMQP_EXCHANGE_TYPE: "direct"

      OUTPUT_AMQP_KEY: "benthos-key"

启动pipeline

效果

• 访问(发送消息)

curl http://localhost:8080/post -d "[email protected]"

• 效果
  
  

配置问题说明

目前发现streamsets rabbit 配置必须包含一个vhost 不能使用默认的。

参考资料

https://github.com/rongfengliang/streamsets-demos
https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Origins/RabbitMQ.html#concept_dyg_lq1_h5
https://github.com/Jeffail/benthos