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

2. 应用配置

在 application.properties 中指定连接信息：

# 服务器端口
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. 代码实现

定义实体类 Product 来映射表结构，注意字段类型与 Hive 类型的对应：

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 and Setter methods
    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; }

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

使用 JdbcTemplate 封装数据访问层，简化 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 Product getProductById(Long id) {
        String sql = "SELECT * FROM product WHERE id = ?";
        return jdbcTemplate.queryForObject(sql, new Object[]{id}, (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());
    }

    public void updateProduct(Product product) {
        String sql = "UPDATE product SET product_id = ?, product_name = ?, price = ?, sales = ? WHERE id = ?";
        jdbcTemplate.update(sql, product.getProductId(), product.getProductName(), product.getPrice(), product.getSales(), product.getId());
    }

    public void deleteProduct(Long id) {
        String sql = "DELETE FROM product WHERE id = ?";
        jdbcTemplate.update(sql, id);
    }
}

配合 Service 层和 Controller 层暴露 REST API，即可完成基础的数据交互功能。

集成 Apache Spark 进行 ETL 处理

当数据量较大或需要复杂转换逻辑时，Spark 是更好的选择。Spring Boot 可以嵌入 Spark Session 来执行任务。

1. 依赖与配置

引入 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 Master 地址和应用名称：

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

2. ETL 任务实现

创建一个组件类来管理 SparkSession 的生命周期和执行逻辑：

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

            // 写入目标数据仓库
            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();
        }
    }
}

3. 任务调度

利用 Spring 的定时任务注解 @Scheduled 可以轻松实现自动化调度：

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

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

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

启动类需开启调度支持：

@SpringBootApplication
@EnableScheduling
public class ETLApplication {
    public static void main(String[] args) {
        SpringApplication.run(ETLApplication.class, args);
    }
}

总结

将 Spring Boot 与数据仓库及 ETL 工具集成，关键在于理解各自的角色：Spring Boot 负责业务逻辑与接口暴露，Hive 负责数据存储与分析，Spark 负责大规模数据处理。在实际开发中，根据数据量和实时性要求选择合适的组合，并注意资源管理和异常处理，就能构建出稳健的数据处理管道。