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

2. 应用配置

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

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. 数据访问层实现

定义实体类 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 and Setter methods omitted for brevity
    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 封装数据库操作：

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

    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

对于更复杂的数据处理需求，Spark 提供了强大的分布式计算能力。我们可以在 Spring Boot 中嵌入 SparkSession 来执行 ETL 任务。

1. 依赖与配置

添加 Spark 核心及 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 运行参数：

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

2. ETL 任务实现

创建一个组件类来管理 Spark 会话和数据处理逻辑：

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

    @Scheduled(cron = "0 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 工具的核心方法。无论是使用 Hive JDBC 进行轻量级数据交互，还是引入 Spark 处理大规模数据清洗，关键在于合理选择技术栈并规范代码结构。实际开发中，可根据业务规模灵活组合这些组件，构建高效可靠的数据管道。