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

配置文件 application.properties 中需指定 Hive 的连接地址与认证信息：

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=

接下来是数据访问层。使用 JdbcTemplate 配合 RowMapper 可以方便地将查询结果映射为实体对象。注意字段名与数据库列名的对应关系，例如 product_id 映射到 productId。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.stereotype.Repository;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;

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

    public List<Product> getAllProducts() {
        String sql = "SELECT * FROM product";
        return jdbcTemplate.query(sql, new RowMapper<Product>() {
            @Override
            public Product mapRow(ResultSet rs, int rowNum) throws SQLException {
                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;
            }
        });
    }

    // 其他 CRUD 方法...
}

业务层与控制器类遵循标准分层模式，通过 RESTful 接口暴露数据操作能力。启动类无需特殊配置，Spring Boot 会自动扫描组件。

集成 Apache Spark

对于更复杂的 ETL 场景，直接嵌入 Spark Session 是常见做法。这允许我们在应用内部执行数据清洗逻辑，而无需依赖外部调度器。

依赖方面，除了 Web 模块，还需引入 Spark Core 和 SQL 模块：

<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 运行环境，通常设置为本地模式或集群模式：

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

核心逻辑在于构建 SparkSession，读取源数据（如 CSV），进行过滤转换，最后写入目标数据仓库。这里以写入 Hive 为例：

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.util.Properties;

@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");

        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));

        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();
    }
}

为了便于触发，我们可以结合 Spring 的定时任务功能，或者提供一个手动触发的 API 接口。启用 @EnableScheduling 后，即可定义 Cron 表达式实现每日凌晨自动执行。

实际应用场景

这种架构适用于多种业务场景，例如产品信息的批量更新、用户行为日志的清洗入库，或是销售报表的自动化生成。关键在于根据数据量选择合适的数据源与计算引擎，并通过 Spring Boot 统一管控生命周期。

结语

通过上述实践，我们实现了 Spring Boot 对 Hive 的直接访问以及对 Spark ETL 任务的嵌入式调度。在实际生产中，建议将重计算任务剥离至独立集群，利用 Airflow 等工具管理依赖，以保证主服务的稳定性。