Spark与RabbitMQ在Ubuntu的消息队列集成

发布时间：2024-10-21 17:20:42 作者：小樊
来源：亿速云阅读：82

在Ubuntu上集成Spark和RabbitMQ以实现消息队列处理，可以按照以下步骤进行：

安装RabbitMQ

更新软件包列表：
```
sudo apt update
```
安装RabbitMQ服务器：
```
sudo apt install rabbitmq-server
```
启动RabbitMQ服务：
```
sudo systemctl start rabbitmq-server
```
设置RabbitMQ开机自启动：
```
sudo systemctl enable rabbitmq-server
```
验证RabbitMQ服务状态：
```
sudo systemctl status rabbitmq-server
```

安装Spark

下载Spark：

wget https://downloads.apache.org/spark/spark-3.2.0/spark-3.2.0-bin-hadoop3.2.tgz

解压Spark：
```
tar -xzf spark-3.2.0-bin-hadoop3.2.tgz
```

设置Spark环境变量：编辑~/.bashrc文件，添加以下内容：

export SPARK_HOME=/path/to/spark-3.2.0-bin-hadoop3.2
export PATH=$PATH:$SPARK_HOME/bin

保存文件并运行：

source ~/.bashrc

验证Spark安装：
```
spark-submit --version
```

配置RabbitMQ与Spark集成

安装RabbitMQ Java客户端库：
```
sudo apt install librabbitmq-java
```

在Spark项目中添加RabbitMQ依赖：在pom.xml文件中添加以下依赖：

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.14.2</version>
</dependency>

编写Spark应用程序：创建一个Java文件，例如RabbitMQSparkApp.java，并编写以下代码：

import com.rabbitmq.client.*;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import scala.Tuple2;

public class RabbitMQSparkApp {

    public static void main(String[] args) throws Exception {
        // 创建Spark配置
        SparkConf conf = new SparkConf().setAppName("RabbitMQSparkApp").setMaster("local[*]");

        // 创建Spark上下文
        JavaSparkContext sc = new JavaSparkContext(conf);

        // 创建RabbitMQ连接
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        // 声明队列
        channel.queueDeclare("spark_queue", false, false, false, null);

        // 读取队列消息
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("Received message: " + message);

                // 处理消息并发送到另一个队列
                String[] parts = message.split(",");
                String processedMessage = parts[0] + "_" + parts[1];
                channel.basicPublish("", "processed_queue", properties, processedMessage.getBytes());
            }
        };
        channel.basicConsume("spark_queue", true, consumer);
    }
}

编译并运行Spark应用程序：

mvn clean package
spark-submit --class RabbitMQSparkApp --master local[*] target/dependency/spark-examples.jar

启动另一个消费者处理已处理的消息

创建一个新的Java文件，例如ProcessedMessageApp.java，并编写以下代码：

import com.rabbitmq.client.*;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import scala.Tuple2;

public class ProcessedMessageApp {

    public static void main(String[] args) throws Exception {
        // 创建Spark配置
        SparkConf conf = new SparkConf().setAppName("ProcessedMessageApp").setMaster("local[*]");

        // 创建Spark上下文
        JavaSparkContext sc = new JavaSparkContext(conf);

        // 创建RabbitMQ连接
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        // 声明队列
        channel.queueDeclare("processed_queue", false, false, false, null);

        // 读取队列消息
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("Received processed message: " + message);
            }
        };
        channel.basicConsume("processed_queue", true, consumer);
    }
}