使用testcontainers测试kafka和spark

Kafka是一个分布式流处理平台，常用于构建高吞吐量、可扩展的实时数据流应用程序。Spark是一个快速、通用的大数据处理引擎，提供了强大的批处理、交互式查询和流处理功能。

在使用testcontainers进行kafka和spark的测试时，可以按照以下步骤进行：

安装Docker：testcontainers是一个基于Docker的Java库，因此需要先安装Docker。
引入testcontainers依赖：在项目的构建文件中引入testcontainers的依赖，例如使用Maven的话，可以在pom.xml文件中添加以下依赖：

<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>kafka</artifactId>
    <version>1.16.0</version>
    <scope>test</scope>
</dependency>

<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>spark</artifactId>
    <version>1.16.0</version>
    <scope>test</scope>
</dependency>

编写测试代码：创建一个测试类，并在测试方法中编写测试代码。以下是一个示例：

import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.KafkaContainer;
import org.testcontainers.containers.SparkContainer;
import org.testcontainers.utility.DockerImageName;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeoutException;

import static org.junit.jupiter.api.Assertions.assertEquals;

public class KafkaSparkTest {

    private KafkaContainer kafkaContainer;
    private SparkContainer sparkContainer;

    @BeforeEach
    public void setup() {
        kafkaContainer = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.5.1"));
        kafkaContainer.start();

        sparkContainer = new SparkContainer(DockerImageName.parse("apache/spark:3.0.1"));
        sparkContainer.dependsOn(kafkaContainer);
        sparkContainer.start();
    }

    @AfterEach
    public void teardown() {
        sparkContainer.stop();
        kafkaContainer.stop();
    }

    @Test
    public void testKafkaAndSpark() throws TimeoutException, StreamingQueryException {
        // 创建Kafka生产者并发送消息
        kafkaContainer.getKafkaProducer().send(new ProducerRecord<>("test-topic", "key", "value"));

        // 创建SparkSession
        SparkSession sparkSession = SparkSession.builder()
                .appName("KafkaSparkTest")
                .master(sparkContainer.getSparkMasterUrl())
                .getOrCreate();

        // 从Kafka主题中读取数据
        Dataset<Row> kafkaData = sparkSession
                .readStream()
                .format("kafka")
                .option("kafka.bootstrap.servers", kafkaContainer.getBootstrapServers())
                .option("subscribe", "test-topic")
                .load();

        // 处理数据
        Dataset<Row> processedData = kafkaData.selectExpr("CAST(value AS STRING)");

        // 启动流处理查询
        StreamingQuery query = processedData.writeStream()
                .format("console")
                .start();

        // 等待查询完成
        query.awaitTermination(5000);

        // 检查输出结果
        List<String> output = sparkContainer.getLogs();
        List<String> expectedOutput = Arrays.asList("value");
        assertEquals(expectedOutput, output);
    }
}

在上述示例中，我们使用了testcontainers提供的KafkaContainer和SparkContainer来启动Kafka和Spark容器。在测试方法中，我们创建了一个Kafka生产者并发送一条消息到指定的主题，然后使用SparkSession从Kafka主题中读取数据，并对数据进行处理，最后将结果输出到控制台。我们还使用了assertions来验证输出结果是否符合预期。