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

2. 配置文件编写

在 application.properties 中指定数据源信息。注意，Hive 默认端口通常是 10000，且需确保网络可达。

# 服务端口
server.port=8080

# Hive 数据源配置
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. 数据访问层实现

使用 JdbcTemplate 是最轻量级的选择。我们定义一个实体类 Product 来映射表结构。

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

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

    // Getter 和 Setter 方法省略，保持代码简洁
    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 接口，封装 SQL 逻辑：

@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 方法类似，此处省略
}

配合 Service 层和 Controller 层暴露 REST API，即可完成基础 CRUD 功能。

Spring Boot 与 ETL 工具集成

当数据量增大或需要实时处理时，Spark 是更好的选择。Spring Boot 可以嵌入 Spark Session 来执行 ETL 任务。

1. 引入 Spark 依赖

相比 Hive，Spark 依赖包体积较大，需注意版本兼容性（特别是 Scala 版本）。

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    
    <!-- Spark Core -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.12</artifactId>
        <version>3.1.2</version>
    </dependency>
    
    <!-- Spark SQL -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.12</artifactId>
        <version>3.1.2</version>
    </dependency>
</dependencies>

2. 配置 Spark 环境

server.port=8080
spark.master=local[*]
spark.app.name=ETLExample

3. 编写 ETL 任务

创建一个组件类来管理 SparkSession 的生命周期。这里演示从 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();

        try {
            // 读取源数据
            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);
        } finally {
            sparkSession.stop();
        }
    }
}

4. 任务调度与触发

为了自动化运行，我们可以结合 Spring 的定时任务注解，或者提供一个 HTTP 接口手动触发。

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

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

    // 手动触发接口
    public void runETLNow() {
        etlJob.runETL();
    }
}

Controller 层只需简单调用即可：

@RestController
@RequestMapping("/api/etl")
public class ETLController {
    @Autowired
    private ETLScheduler etlScheduler;

    @PostMapping("/run")
    public String runETL() {
        etlScheduler.runETL();
        return "ETL 任务已启动";
    }
}

实际应用场景

这种集成模式在实际开发中非常灵活，常见的场景包括：

• 商品信息同步：将电商交易系统的订单数据定期同步至数仓进行分析。
• 用户画像更新：根据用户行为日志，每日更新用户标签体系。
• 销售报表生成：聚合多源销售数据，生成管理层日报。

注意事项：

1. 资源隔离：Spark 任务消耗内存较大，建议在生产环境中独立部署集群，避免影响 Spring Boot 主业务线程。
2. 异常处理：ETL 任务失败需要有告警机制，不能静默失败。
3. 数据一致性：在写入 Hive 前做好事务控制或幂等性设计。

总结

通过 Spring Boot 整合 Hive 和 Spark，我们能够以较低的成本构建起现代化的数据处理链路。对于中小规模数据，直接利用 JDBC 连接 Hive 足够高效；而对于大规模离线或实时计算，嵌入 Spark Session 则提供了强大的扩展能力。关键在于根据业务场景选择合适的工具组合，并做好资源管理与错误处理。