如何在 DAG 中使用 Apache Apex Malhar RabbitMQ 运算符

Question

我有一个 Apache Apex 应用程序 DAG，它从队列中读取 RabbitMQ 消息。我应该使用哪个 Apache Apex Malhar 运算符？有几个运算符，但不清楚使用哪一个以及如何使用它。

Answer 1

你看过https://github.com/apache/apex-malhar/tree/master/contrib/src/main/java/com/datatorrent/contrib/rabbitmq 吗？ https://github.com/apache/apex-malhar/tree/master/contrib/src/test/java/com/datatorrent/contrib/rabbitmq 中也有测试显示如何使用操作符

Answer 2

https://github.com/apache/apex-malhar/blob/master/contrib/src/main/java/com/datatorrent/contrib/rabbitmq/AbstractRabbitMQInputOperator.java

这是主要的操作符代码，其中元组类型是泛型参数，emitTuple() 是子类需要实现的抽象方法。

AbstractSinglePortRabbitMQInputOperator 是一个简单的子类，它提供单个输出端口并使用另一个抽象方法 getTuple() 来实现 emitTuple()，该方法需要在其子类中实现。

Sanjay 指出的测试显示了如何使用这些类。

Answer 3

我在查找如何将消息从 RabbitMQ 读取到 Apache Apex 时也遇到了问题。在 Sanjay 的回答 (https://stackoverflow.com/a/42210636/2350644) 提供的链接的帮助下，我终于设法让它运行起来。以下是它如何协同工作：

1。设置 RabbitMQ 服务器

这里描述了很多安装 RabbitMQ 的方法：https://www.rabbitmq.com/download.html
对我来说最简单的方法是使用docker（参见：https://store.docker.com/images/rabbitmq）

docker pull rabbitmq

docker run -d --hostname my-rabbit --name some-rabbit -p 5672:5672 -p 15672:15672 rabbitmq:3-management  

要检查 RabbitMQ 是否正常工作，请打开浏览器并导航到：http://localhost:15672/。您应该看到 RabbitMQ 的管理页面。

2。编写生产者程序

要向队列发送消息，您可以编写一个简单的 JAVA 程序，如下所示：

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.util.ArrayList;

public class Send {
    private final static String EXCHANGE = "myExchange";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.FANOUT);

        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName, EXCHANGE, "");

        List<String> messages = Arrays.asList("Hello", "World", "!");
        for (String msg : messages) {
            channel.basicPublish(EXCHANGE, "", null, msg.getBytes("UTF-8"));
            System.out.println(" [x] Sent '" + msg + "'");
        }

        channel.close();
        connection.close();
    }
}

如果您执行 JAVA 程序，您应该会在 RabbitMQ 的管理 UI 中看到一些输出。

3。实施示例 Apex 应用程序

3.1 引导示例 apex 应用程序

关注apex官方文档http://docs.datatorrent.com/beginner/

3.2 向 pom.xml 添加额外的依赖项

要使用malhar 提供的类，请添加以下依赖项：

<dependency>
  <groupId>org.apache.apex</groupId>
  <artifactId>malhar-contrib</artifactId>
  <version>3.7.0</version>
</dependency>
<dependency>
  <groupId>com.rabbitmq</groupId>
  <artifactId>amqp-client</artifactId>
  <version>4.2.0</version>
</dependency>

3.3 创建消费者

我们首先需要创建一个InputOperator，它使用来自apex-malhar 的可用代码来使用来自RabbitMQ 的消息。

import com.datatorrent.contrib.rabbitmq.AbstractSinglePortRabbitMQInputOperator;

public class MyRabbitMQInputOperator extends AbstractSinglePortRabbitMQInputOperator<String> {
    @Override
    public String getTuple(byte[] message) {
        return new String(message);
    }
}

您只需覆盖getTuple() 方法。在这种情况下，我们只是返回从 RabbitMQ 接收到的消息。

3.4 设置 Apex DAG

为了测试应用程序，我们只需添加一个使用来自 RabbitMQ 的数据的InputOperator（我们之前实现的MyRabbitMQInputOperator）和一个打印接收到的消息的ConsoleOutputOperator。

import com.rabbitmq.client.BuiltinExchangeType;
import org.apache.hadoop.conf.Configuration;

import com.datatorrent.api.annotation.ApplicationAnnotation;
import com.datatorrent.api.StreamingApplication;
import com.datatorrent.api.DAG;
import com.datatorrent.api.DAG.Locality;
import com.datatorrent.lib.io.ConsoleOutputOperator;

@ApplicationAnnotation(name="MyFirstApplication")
public class Application implements StreamingApplication
{

  private final static String EXCHANGE = "myExchange";


  @Override
  public void populateDAG(DAG dag, Configuration conf)
  {

    MyRabbitMQInputOperator consumer = dag.addOperator("Consumer", new MyRabbitMQInputOperator());
    consumer.setHost("localhost");
    consumer.setExchange(EXCHANGE);
    consumer.setExchangeType(BuiltinExchangeType.FANOUT.getType());

    ConsoleOutputOperator cons = dag.addOperator("console", new ConsoleOutputOperator());

    dag.addStream("myStream", consumer.outputPort, cons.input).setLocality(Locality.CONTAINER_LOCAL);
  }
}

3.5 测试应用程序

为了简单地测试创建的应用程序，我们可以编写一个 UnitTest，因此无需设置 Hadoop/YARN 集群。 在引导应用程序中已经有一个 UnitTest，即我们可以使用的 ApplicationTest.java：

import java.io.IOException;
import javax.validation.ConstraintViolationException;
import org.junit.Assert;
import org.apache.hadoop.conf.Configuration;
import org.junit.Test;
import com.datatorrent.api.LocalMode;

/**
 * Test the DAG declaration in local mode.
 */
public class ApplicationTest {

  @Test
  public void testApplication() throws IOException, Exception {
    try {
      LocalMode lma = LocalMode.newInstance();
      Configuration conf = new Configuration(true);
      //conf.addResource(this.getClass().getResourceAsStream("/META-INF/properties.xml"));
      lma.prepareDAG(new Application(), conf);
      LocalMode.Controller lc = lma.getController();
      lc.run(10000); // runs for 10 seconds and quits
    } catch (ConstraintViolationException e) {
      Assert.fail("constraint violations: " + e.getConstraintViolations());
    }
  }

}

由于我们不需要此应用程序的任何属性，因此该文件中唯一更改的是取消注释该行：

conf.addResource(this.getClass().getResourceAsStream("/META-INF/properties.xml"));

如果您执行ApplicationTest.java 并使用2. 中描述的Producer 程序向RabbitMQ 发送消息，则测试应输出所有消息。 您可能需要增加测试时间才能看到所有消息（目前设置为 10 秒）。