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. 实体类与 Repository

定义数据模型后，通过 RowMapper 映射结果集。这里以产品表为例。

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; }
    // ... 其他字段 getter/setter
}

Repository 层封装 SQL 逻辑，利用 JdbcTemplate 执行查询与更新。

@Repository
public class ProductRepository {
    @Autowired
    private JdbcTemplate jdbcTemplate;

    public List<Product> getAllProducts() {
        String sql = "SELECT * FROM product";
        return jdbcTemplate.query(sql, (rs, rowNum) -> {
            Product p = new Product();
            p.setId(rs.getLong("id"));
            p.setProductId(rs.getString("product_id"));
            p.setProductName(rs.getString("product_name"));
            p.setPrice(rs.getDouble("price"));
            p.setSales(rs.getInt("sales"));
            return p;
        });
    }

    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 等 CRUD 方法类似实现
}

配合 Service 层与 Controller 暴露 REST API，即可完成对 Hive 数据的操作。

集成 Apache Spark：ETL 任务编排

对于更复杂的数据清洗与转换，Spark 是更好的选择。Spring Boot 可以通过嵌入 SparkSession 来运行 ETL 作业。

1. 依赖与配置

<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 任务实现

创建一个组件类来处理数据流转。这里演示从 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.filter(sourceData.col("sales").gt(100))
                    .select(sourceData.col("id"), sourceData.col("product_id"), 
                            sourceData.col("product_name"), sourceData.col("price"), sourceData.col("sales"));

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

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 能够无缝衔接大数据生态。集成 Hive 侧重于数据持久化与查询，而集成 Spark 则强化了计算与转换能力。在实际项目中，建议根据数据量级选择合适的工具组合，并注意依赖版本的兼容性，避免运行时出现类冲突或连接超时问题。