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

Ne0inhk

— 10 min read
Spring Boot 数据仓库与ETL工具集成

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

在这里插入图片描述
26.1 学习目标与重点提示

学习目标：掌握Spring Boot数据仓库与ETL工具集成的核心概念与使用方法，包括数据仓库的定义与特点、ETL工具的定义与特点、Spring Boot与数据仓库的集成、Spring Boot与ETL工具的集成、Spring Boot的实际应用场景，学会在实际开发中处理数据仓库与ETL工具集成问题。
重点：数据仓库的定义与特点、ETL工具的定义与特点、Spring Boot与数据仓库的集成、Spring Boot与ETL工具的集成、Spring Boot的实际应用场景。

26.2 数据仓库与ETL工具概述

数据仓库与ETL工具是Java开发中的重要组件。

26.2.1 数据仓库的定义

定义：数据仓库是一种用于存储和管理大量结构化数据的数据库系统，用于支持企业级数据分析和决策。
作用：

  • 提供统一的数据存储。
  • 支持复杂的数据分析。
  • 提高决策效率。

常见的数据仓库：

  • Apache Hive：Apache Hive是一种基于Hadoop的数据仓库工具。
  • Apache HBase：Apache HBase是一种基于Hadoop的列式数据库。
  • Amazon Redshift：Amazon Redshift是一种基于云计算的数据仓库。
  • Google BigQuery：Google BigQuery是一种基于云计算的数据仓库。

✅ 结论：数据仓库是一种用于存储和管理大量结构化数据的数据库系统，作用是提供统一的数据存储、支持复杂的数据分析、提高决策效率。

26.2.2 ETL工具的定义

定义：ETL工具是一种用于数据抽取（Extract）、转换（Transform）和加载（Load）的工具，用于将数据从源系统导入到数据仓库。
作用：

  • 实现数据的抽取。
  • 实现数据的转换。
  • 实现数据的加载。

常见的ETL工具：

  • Apache Spark：Apache Spark是一种开源的分布式计算框架，支持ETL操作。
  • Apache Flink：Apache Flink是一种开源的流处理框架，支持ETL操作。
  • Apache Airflow：Apache Airflow是一种开源的调度工具，用于调度ETL任务。
  • Talend：Talend是一种开源的ETL工具。

✅ 结论：ETL工具是一种用于数据抽取、转换和加载的工具，作用是实现数据的抽取、转换、加载。

26.3 Spring Boot与数据仓库的集成

Spring Boot与数据仓库的集成是Java开发中的重要内容。

26.3.1 集成Apache Hive的步骤

定义：集成Apache Hive的步骤是指使用Spring Boot与Apache Hive集成的方法。
步骤：

  1. 创建Spring Boot项目。
  2. 添加所需的依赖。
  3. 配置Apache Hive。
  4. 创建数据访问层。
  5. 创建业务层。
  6. 创建控制器类。
  7. 测试应用。

示例：
pom.xml文件中的依赖：

<dependencies><!-- Web依赖 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- Hive依赖 --><dependency><groupId>org.apache.hive</groupId><artifactId>hive-jdbc</artifactId><version>3.1.2</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>3.3.1</version></dependency><!-- 测试依赖 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies>

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=

实体类：

publicclassProduct{privateLong id;privateString productId;privateString productName;privatedouble price;privateint sales;publicProduct(){}publicProduct(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方法publicLonggetId(){return id;}publicvoidsetId(Long id){this.id = id;}publicStringgetProductId(){return productId;}publicvoidsetProductId(String productId){this.productId = productId;}publicStringgetProductName(){return productName;}publicvoidsetProductName(String productName){this.productName = productName;}publicdoublegetPrice(){return price;}publicvoidsetPrice(double price){this.price = price;}publicintgetSales(){return sales;}publicvoidsetSales(int sales){this.sales = sales;}@OverridepublicStringtoString(){return"Product{"+"id="+ id +",+ productId +'\''+",+ productName +'\''+", price="+ price +", sales="+ sales +'}';}}

Repository接口：

importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.jdbc.core.JdbcTemplate;importorg.springframework.jdbc.core.RowMapper;importorg.springframework.stereotype.Repository;importjava.sql.ResultSet;importjava.sql.SQLException;importjava.util.List;@RepositorypublicclassProductRepository{@AutowiredprivateJdbcTemplate jdbcTemplate;publicList<Product>getAllProducts(){String sql ="SELECT * FROM product";return jdbcTemplate.query(sql,newRowMapper<Product>(){@OverridepublicProductmapRow(ResultSet rs,int rowNum)throwsSQLException{Product product =newProduct(); 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;}});}publicProductgetProductById(Long id){String sql ="SELECT * FROM product WHERE id = ?";return jdbcTemplate.queryForObject(sql,newObject[]{id},newRowMapper<Product>(){@OverridepublicProductmapRow(ResultSet rs,int rowNum)throwsSQLException{Product product =newProduct(); 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;}});}publicvoidaddProduct(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());}publicvoidupdateProduct(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());}publicvoiddeleteProduct(Long id){String sql ="DELETE FROM product WHERE id = ?"; jdbcTemplate.update(sql, id);}}

Service类：

importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Service;importjava.util.List;@ServicepublicclassProductService{@AutowiredprivateProductRepository productRepository;publicList<Product>getAllProducts(){return productRepository.getAllProducts();}publicProductgetProductById(Long id){return productRepository.getProductById(id);}publicvoidaddProduct(Product product){ productRepository.addProduct(product);}publicvoidupdateProduct(Product product){ productRepository.updateProduct(product);}publicvoiddeleteProduct(Long id){ productRepository.deleteProduct(id);}}

控制器类：

importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.*;importjava.util.List;@RestController@RequestMapping("/api/products")publicclassProductController{@AutowiredprivateProductService productService;@GetMapping("/")publicList<Product>getAllProducts(){return productService.getAllProducts();}@GetMapping("/{id}")publicProductgetProductById(@PathVariableLong id){return productService.getProductById(id);}@PostMapping("/")publicvoidaddProduct(@RequestBodyProduct product){ productService.addProduct(product);}@PutMapping("/{id}")publicvoidupdateProduct(@PathVariableLong id,@RequestBodyProduct product){ product.setId(id); productService.updateProduct(product);}@DeleteMapping("/{id}")publicvoiddeleteProduct(@PathVariableLong id){ productService.deleteProduct(id);}}

应用启动类：

importorg.springframework.boot.SpringApplication;importorg.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplicationpublicclassHiveApplication{publicstaticvoidmain(String[] args){SpringApplication.run(HiveApplication.class, args);}}

测试类：

importorg.junit.jupiter.api.Test;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.boot.test.context.SpringBootTest;importorg.springframework.boot.test.web.client.TestRestTemplate;importorg.springframework.boot.web.server.LocalServerPort;importstaticorg.assertj.core.api.Assertions.assertThat;@SpringBootTest(webEnvironment =SpringBootTest.WebEnvironment.RANDOM_PORT)classHiveApplicationTests{@LocalServerPortprivateint port;@AutowiredprivateTestRestTemplate restTemplate;@TestvoidcontextLoads(){}@TestvoidtestGetAllProducts(){List<Product> products = restTemplate.getForObject("http://localhost:"+ port +"/api/products/",List.class);assertThat(products).isNotNull();assertThat(products.size()).isGreaterThanOrEqualTo(0);}@TestvoidtestAddProduct(){Product product =newProduct(null,"P001","手机",1000.0,100); restTemplate.postForObject("http://localhost:"+ port +"/api/products/", product,Product.class);List<Product> products = restTemplate.getForObject("http://localhost:"+ port +"/api/products/",List.class);assertThat(products).isNotNull();assertThat(products.size()).isGreaterThanOrEqualTo(1);}}

✅ 结论：集成Apache Hive的步骤包括创建Spring Boot项目、添加所需的依赖、配置Apache Hive、创建数据访问层、创建业务层、创建控制器类、测试应用。

26.4 Spring Boot与ETL工具的集成

Spring Boot与ETL工具的集成是Java开发中的重要内容。

26.4.1 集成Apache Spark的步骤

定义：集成Apache Spark的步骤是指使用Spring Boot与Apache Spark集成的方法。
步骤：

  1. 创建Spring Boot项目。
  2. 添加所需的依赖。
  3. 配置Apache Spark。
  4. 创建ETL任务。
  5. 测试应用。

示例：
pom.xml文件中的依赖：

<dependencies><!-- Web依赖 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- Spark依赖 --><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><!-- 测试依赖 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies>

application.properties文件中的配置：

# 服务器端口 server.port=8080 # Spark配置 spark.master=local[*] spark.app.name=ETLExample

ETL任务类：

importorg.apache.spark.sql.Dataset;importorg.apache.spark.sql.Row;importorg.apache.spark.sql.SparkSession;importorg.springframework.beans.factory.annotation.Value;importorg.springframework.stereotype.Component;importjava.util.Properties;@ComponentpublicclassETLJob{@Value("${spark.master}")privateString master;@Value("${spark.app.name}")privateString appName;publicvoidrunETL(){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");// 数据转换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")>100);// 写入目标数据Properties connectionProperties =newProperties(); connectionProperties.put("user","hive"); connectionProperties.put("password",""); transformedData.write().mode("overwrite").jdbc("jdbc:hive2://localhost:10000/default","transformed_product", connectionProperties); sparkSession.stop();}}

控制器类：

importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.web.bind.annotation.*;@RestController@RequestMapping("/api/etl")publicclassETLController{@AutowiredprivateETLScheduler etlScheduler;@PostMapping("/run")publicStringrunETL(){ etlScheduler.runETL();return"ETL任务已启动";}}

调度器类：

importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.scheduling.annotation.Scheduled;importorg.springframework.stereotype.Component;@ComponentpublicclassETLScheduler{@AutowiredprivateETLJob etlJob;@Scheduled(cron ="0 0 0 * * ?")// 每天凌晨0点执行publicvoidrunETL(){ etlJob.runETL();}publicvoidrunETLNow(){ etlJob.runETL();}}

应用启动类：

importorg.springframework.boot.SpringApplication;importorg.springframework.boot.autoconfigure.SpringBootApplication;importorg.springframework.scheduling.annotation.EnableScheduling;@SpringBootApplication@EnableSchedulingpublicclassETLApplication{publicstaticvoidmain(String[] args){SpringApplication.run(ETLApplication.class, args);}}

测试类：

importorg.junit.jupiter.api.Test;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.boot.test.context.SpringBootTest;importorg.springframework.boot.test.web.client.TestRestTemplate;importorg.springframework.boot.web.server.LocalServerPort;importstaticorg.assertj.core.api.Assertions.assertThat;@SpringBootTest(webEnvironment =SpringBootTest.WebEnvironment.RANDOM_PORT)classETLApplicationTests{@LocalServerPortprivateint port;@AutowiredprivateTestRestTemplate restTemplate;@TestvoidcontextLoads(){}@TestvoidtestRunETL(){String response = restTemplate.postForObject("http://localhost:"+ port +"/api/etl/run",null,String.class);assertThat(response).contains("ETL任务已启动");}}

✅ 结论：集成Apache Spark的步骤包括创建Spring Boot项目、添加所需的依赖、配置Apache Spark、创建ETL任务、测试应用。

26.5 Spring Boot的实际应用场景

在实际开发中，Spring Boot数据仓库与ETL工具集成的应用场景非常广泛，如：

  • 实现产品信息的ETL任务。
  • 实现用户信息的ETL任务。
  • 实现订单信息的ETL任务。
  • 实现销售数据的ETL任务。

示例：

importorg.apache.spark.sql.Dataset;importorg.apache.spark.sql.Row;importorg.apache.spark.sql.SparkSession;importorg.springframework.beans.factory.annotation.Value;importorg.springframework.stereotype.Component;importjava.util.Properties;@ComponentclassETLJob{@Value("${spark.master}")privateString master;@Value("${spark.app.name}")privateString appName;publicvoidrunETL(){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");// 数据转换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")>100);// 写入目标数据Properties connectionProperties =newProperties(); connectionProperties.put("user","hive"); connectionProperties.put("password",""); transformedData.write().mode("overwrite").jdbc("jdbc:hive2://localhost:10000/default","transformed_product", connectionProperties); sparkSession.stop();}}@RestController@RequestMapping("/api/etl")classETLController{@AutowiredprivateETLScheduler etlScheduler;@PostMapping("/run")publicStringrunETL(){ etlScheduler.runETL();return"ETL任务已启动";}}@ComponentclassETLScheduler{@AutowiredprivateETLJob etlJob;@Scheduled(cron ="0 0 0 * * ?")// 每天凌晨0点执行publicvoidrunETL(){ etlJob.runETL();}publicvoidrunETLNow(){ etlJob.runETL();}}@SpringBootApplication@EnableSchedulingpublicclassETLApplication{publicstaticvoidmain(String[] args){SpringApplication.run(ETLApplication.class, args);}}// 测试类@SpringBootTest(webEnvironment =SpringBootTest.WebEnvironment.RANDOM_PORT)classETLApplicationTests{@LocalServerPortprivateint port;@AutowiredprivateTestRestTemplate restTemplate;@TestvoidcontextLoads(){}@TestvoidtestRunETL(){String response = restTemplate.postForObject("http://localhost:"+ port +"/api/etl/run",null,String.class);assertThat(response).contains("ETL任务已启动");}}

输出结果：

  • 访问http://localhost:8080/api/etl/run：启动ETL任务。
  • 控制台输出：ETL任务已启动。

✅ 结论：在实际开发中，Spring Boot数据仓库与ETL工具集成的应用场景非常广泛，需要根据实际问题选择合适的数据仓库和ETL工具。

总结

本章我们学习了Spring Boot数据仓库与ETL工具集成，包括数据仓库的定义与特点、ETL工具的定义与特点、Spring Boot与数据仓库的集成、Spring Boot与ETL工具的集成、Spring Boot的实际应用场景，学会了在实际开发中处理数据仓库与ETL工具集成问题。其中，数据仓库的定义与特点、ETL工具的定义与特点、Spring Boot与数据仓库的集成、Spring Boot与ETL工具的集成、Spring Boot的实际应用场景是本章的重点内容。从下一章开始，我们将学习Spring Boot的其他组件、微服务等内容。

Read more

MySQL 动态分区管理：自动化与优化实践

MySQL 动态分区管理：自动化与优化实践

个人名片 🎓作者简介：java领域优质创作者 🌐个人主页：码农阿豪 📞工作室：新空间代码工作室（提供各种软件服务) 💌个人邮箱：[[email protected]] 📱个人微信：15279484656 🌐个人导航网站：www.forff.top 💡座右铭：总有人要赢。为什么不能是我呢？ * 专栏导航： 码农阿豪系列专栏导航 面试专栏：收集了java相关高频面试题，面试实战总结🍻🎉🖥️ Spring5系列专栏：整理了Spring5重要知识点与实战演练，有案例可直接使用🚀🔧💻 Redis专栏：Redis从零到一学习分享，经验总结，案例实战💐📝💡 全栈系列专栏：海纳百川有容乃大，可能你想要的东西里面都有🤸🌱🚀 目录 * MySQL 动态分区管理：自动化与优化实践 * 一、分区的基本概念 * 二、动态分区的需求 * 三、使用存储过程动态创建分区 * 四、使用事件调度器自动化分区管理 * 五、避免分区冲突

By Ne0inhk

AD的快捷键方式

新手生存必备（第一周必须掌握） 这些是让你能顺畅完成第一张原理图的基础操作。 通用操作 快捷键功能使用频率Ctrl + S保存当前文件⭐⭐⭐⭐⭐Ctrl + 鼠标滚轮放大/缩小视图⭐⭐⭐⭐⭐鼠标滚轮按下拖动平移画布⭐⭐⭐⭐⭐空格 Space旋转元件/对象（放置时或选中后）⭐⭐⭐⭐⭐X / Y水平/垂直镜像翻转（放置元件时按）⭐⭐⭐⭐Delete删除选中对象⭐⭐⭐⭐End刷新屏幕（解决显示残影）⭐⭐⭐ 原理图绘制核心 快捷键功能说明P → P放置元件（Place Part）从库中调取元件P → W放置导线（Place Wire）必须用这个！不能用画图直线P → N放置网络标签（Place Net Label）给导线命名，相同名字即连通P → O放置电源端口（Power Port）放置VCC等电源符号P → G放置接地端口（GND Port）Tab放置前编辑属性最重要的键！放置任何东西前先按Tab设置好Ctrl + C / V复制/粘贴Ctrl

By Ne0inhk

Flutter 组件 pair 适配鸿蒙 HarmonyOS 实战：结构化元组治理，构建轻量级双元数据模型与跨层传递架构

欢迎加入开源鸿蒙跨平台社区：https://openharmonycrossplatform.ZEEKLOG.net Flutter 组件 pair 适配鸿蒙 HarmonyOS 实战：结构化元组治理，构建轻量级双元数据模型与跨层传递架构 前言 在鸿蒙（OpenHarmony）生态迈向多维数据感知、涉及高频函数返回值传递、两元坐标互操作及复杂状态标识返回的背景下，如何以最轻量化的方式实现数据的“成对化”封装，已成为提升代码整洁度与系统运行效率的“工程润滑剂”。在鸿蒙设备这类强调 AOT 极致性能与低内存开销的环境下，如果应用为了简单的双元数据（如：经纬度、错误码+消息）而动态创建大量繁琐的单次使用类（POJO），由于由于对象头开销与 GC 压力，极易由于由于“类爆炸”导致内存碎片的堆积。 我们需要一种能够支持强类型泛型、具备不可变属性且无需显式类定义的元组治理方案。 pair 为 Flutter 开发者引入了源自 C++ 与 Java 标准库经典语义的“

By Ne0inhk
Rust异步Web框架Axum的深入原理与高级用法

Rust异步Web框架Axum的深入原理与高级用法

Rust异步Web框架Axum的深入原理与高级用法 一、Axum框架的架构与核心组件 1.1 Axum框架的设计理念 💡Axum是基于Tokio异步运行时的Rust Web框架，由Tokio团队官方维护，具有以下核心设计理念： 1. 模块化与可扩展性：通过中间件、请求提取器和响应映射器等组件，实现高度模块化的架构，允许开发者根据需求灵活组合功能。 2. 类型安全：利用Rust的类型系统确保请求处理逻辑的正确性，减少运行时错误。 3. 异步优先：完全基于Tokio异步运行时，充分利用现代硬件的并发能力。 4. 低门槛：提供简单易用的API，同时保持足够的灵活性，适合不同经验水平的开发者。 1.2 Axum框架的核心组件 1.2.1 请求提取器 请求提取器负责从HTTP请求中提取所需的数据，如路径参数、查询参数、请求体等。Axum提供了多种内置的请求提取器，并允许开发者自定义提取器。 内置请求提取器示例： useaxum::{extract::Path,response::IntoResponse,routing::get,

By Ne0inhk