Spring Boot 集成数据仓库与 ETL 工具实战

Spring Boot 集成数据仓库与 ETL 工具实战

在构建企业级应用时，数据处理往往是核心环节。Spring Boot 作为 Java 生态的基石，如何高效地对接数据仓库（Data Warehouse）和 ETL（Extract, Transform, Load）工具，是提升系统数据吞吐能力的关键。本文将结合实际场景，梳理从概念到代码落地的完整流程。

一、核心概念与目标

数据仓库本质上是为分析而生的数据库系统，它汇聚了来自不同业务线的结构化数据，支持复杂的查询与分析决策。常见的如 Apache Hive、Amazon Redshift 等，它们擅长处理海量历史数据。

ETL 工具则负责数据的流动，将源系统的数据抽取出来，经过清洗转换后加载到目标端。Apache Spark 和 Flink 是目前主流的分布式计算框架，配合 Airflow 调度，能构建稳定的数据管道。

我们的目标是掌握如何在 Spring Boot 中无缝集成这些组件，实现自动化数据流转。

二、Spring Boot 与数据仓库集成：以 Hive 为例

集成 Hive 的核心在于配置正确的 JDBC 驱动和数据源。这里我们以一个产品管理场景为例，展示如何将 Hive 表映射为 Spring 实体。

1. 依赖与配置

首先引入必要的 Maven 依赖。除了基础的 Web 启动器，我们需要 Hive JDBC 驱动以及 Hadoop 公共库来保证兼容性。

<dependencies>
    <!-- Web 依赖 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!-- Hive 依赖 -->
    <dependency>
        <groupId>org.apache.hive</groupId>
        <artifactId>hive-jdbc</artifactId>
        <version>3.1.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
        <version>3.3.1</version>
    </dependency>
    <!-- 测试依赖 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

在 application.properties 中配置连接信息，注意 Hive 默认端口通常为 10000。

server.port=8080
spring.datasource.url=jdbc:hive2://localhost:10000/default
spring.datasource.driver-class-name=org.apache.hive.jdbc.HiveDriver
spring.datasource.username=hive
spring.datasource.password=

2. 数据访问层实现

使用 JdbcTemplate 可以简化 SQL 操作。我们定义一个 Product 实体，并编写 Repository 接口来处理 CRUD 逻辑。

public class Product {
    private Long id;
    private String productId;
    private String productName;
    private double price;
    private int sales;

    // 构造函数、Getter/Setter 省略
    public Product() {}

    public Product(Long id, String productId, String productName, double price, int sales) {
        this.id = id;
        this.productId = productId;
        this.productName = productName;
        this.price = price;
        this.sales = sales;
    }

    @Override
    public String toString() {
        return "Product{" +
                "id=" + id +
                ", productId='" + productId + '\'' +
                ", productName='" + productName + '\'' +
                ", price=" + price +
                ", sales=" + sales +
                '}';
    }
}

Repository 层通过 RowMapper 将结果集映射为对象，这里展示了标准的查询与更新逻辑。

@Repository
public class ProductRepository {
    @Autowired
    private JdbcTemplate jdbcTemplate;

    public List<Product> getAllProducts() {
        String sql = "SELECT * FROM product";
        return jdbcTemplate.query(sql, (rs, rowNum) -> {
            Product product = new Product();
            product.setId(rs.getLong("id"));
            product.setProductId(rs.getString("product_id"));
            product.setProductName(rs.getString("product_name"));
            product.setPrice(rs.getDouble("price"));
            product.setSales(rs.getInt("sales"));
            return product;
        });
    }

    public void addProduct(Product product) {
        String sql = "INSERT INTO product (product_id, product_name, price, sales) VALUES (?, ?, ?, ?)";
        jdbcTemplate.update(sql, product.getProductId(), product.getProductName(), product.getPrice(), product.getSales());
    }
    // update 和 delete 方法类似，此处省略
}

配合 Service 和 Controller 层，即可暴露 RESTful API 供前端调用。

三、Spring Boot 与 ETL 工具集成：以 Spark 为例

当数据量增大或需要复杂转换时，Spark 是更好的选择。Spring Boot 本身不直接运行 Spark 任务，但可以通过嵌入 SparkSession 来触发作业。

1. 项目配置

引入 Spark Core 和 SQL 模块，并在配置文件中指定 Master 地址。

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.12</artifactId>
    <version>3.1.2</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.12</artifactId>
    <version>3.1.2</version>
</dependency>

spark.master=local[*]
spark.app.name=ETLExample

2. 执行 ETL 任务

创建一个组件类来封装 Spark 逻辑。这里演示读取 CSV 文件，过滤销售数据大于 100 的记录，并写入 Hive。

@Component
public class ETLJob {
    @Value("${spark.master}")
    private String master;

    @Value("${spark.app.name}")
    private String appName;

    public void runETL() {
        SparkSession sparkSession = SparkSession.builder()
                .master(master)
                .appName(appName)
                .getOrCreate();

        // 读取源数据
        Dataset<Row> sourceData = sparkSession.read()
                .format("csv")
                .option("header", "true")
                .option("inferSchema", "true")
                .load("src/main/resources/source-data.csv");

        // 数据转换：筛选销量大于 100 的商品
        Dataset<Row> transformedData = sourceData.select(
                sourceData.col("id"),
                sourceData.col("product_id"),
                sourceData.col("product_name"),
                sourceData.col("price"),
                sourceData.col("sales")
        ).filter(sourceData.col("sales").gt(100));

        // 写入目标数据至 Hive
        Properties connectionProperties = new Properties();
        connectionProperties.put("user", "hive");
        connectionProperties.put("password", "");
        transformedData.write().mode("overwrite")
                .jdbc("jdbc:hive2://localhost:10000/default", "transformed_product", connectionProperties);

        sparkSession.stop();
    }
}

3. 定时调度

利用 Spring 的 @Scheduled 注解可以轻松实现每日凌晨自动执行。

@Component
public class ETLScheduler {
    @Autowired
    private ETLJob etlJob;

    @Scheduled(cron = "0 0 0 * * ?") // 每天凌晨 0 点执行
    public void runETL() {
        etlJob.runETL();
    }

    public void runETLNow() {
        etlJob.runETL();
    }
}

启动类需开启调度功能：@EnableScheduling。

四、应用场景与总结

这种架构非常适合处理产品库存同步、用户行为分析日志入库等场景。通过 Spring Boot 提供 API 入口，底层由 Spark 处理重负载计算，Hive 负责持久化存储，既保证了开发效率，又兼顾了大数据处理能力。

实际开发中，建议根据数据规模选择合适的工具：小规模实时交互用 Hive 直连，大规模离线批处理则优先选用 Spark。希望本文提供的代码模板能帮助你在项目中快速搭建起可靠的数据管道。