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

在 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。希望本文提供的代码模板能帮助你在项目中快速搭建起可靠的数据管道。