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

2. 环境配置

在 application.properties 中指定 Hive 的连接信息，确保服务能正确路由到 HiveServer2：

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=

3. 数据访问层实现

定义实体类映射表结构，随后通过 Repository 接口封装 SQL 逻辑。这里使用 JdbcTemplate 配合 RowMapper 进行结果集映射，代码简洁且易于维护。

Product 实体类：

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

    // 构造函数、Getter/Setter 省略，实际开发建议使用 Lombok
    public Product() {}

    public Long getId() { return id; }
    public void setId(Long id) { this.id = id; }
    public String getProductId() { return productId; }
    public void setProductId(String productId) { this.productId = productId; }
    public String getProductName() { return productName; }
    public void setProductName(String productName) { this.productName = productName; }
    public double getPrice() { return price; }
    public void setPrice(double price) { this.price = price; }
    public int getSales() { return sales; }
    public void setSales(int sales) { this.sales = sales; }
}

Repository 接口：

@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());
    }
    
    // updateProduct 和 deleteProduct 方法逻辑类似，此处省略
}

4. 业务层与控制器

Service 层负责事务控制与业务逻辑编排，Controller 层暴露 RESTful API 供前端调用。启动类需添加 @EnableScheduling 以便后续扩展定时任务。

集成 Apache Spark 实现 ETL

对于大规模数据处理，Spark 是更优的选择。Spring Boot 可以嵌入 SparkSession，在应用内部直接执行 ETL 逻辑。

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 任务编写

创建一个 Spring Component 来管理 Spark 生命周期。读取 CSV 源数据，进行过滤转换，最后写入 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 注解实现定时触发，或者通过 Controller 手动触发。这为批处理任务提供了极大的灵活性。

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

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

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

总结

通过上述实践，我们实现了 Spring Boot 与 Hive 的直接交互，以及基于 Spark 的分布式 ETL 流程。在实际项目中，可以根据数据规模选择合适的方案：小规模实时查询适合 Hive JDBC，而海量数据清洗则推荐 Spark 集成。掌握这些集成模式，能帮助开发者在 Java 生态中从容应对复杂的数据架构需求。