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

Venera开源漫画阅读器全攻略：从认知到优化的个性化漫画体验

Venera开源漫画阅读器全攻略：从认知到优化的个性化漫画体验 【免费下载链接】veneraA comic app 项目地址: https://gitcode.com/gh_mirrors/ve/venera 一、认知：快速了解Venera的核心价值 Venera是一款功能全面的开源漫画阅读器，专为漫画爱好者设计。它像一个数字漫画图书馆管理员，不仅能帮你整理本地漫画，还能连接网络资源，让你随时随地享受阅读乐趣。 1.1 3分钟了解核心功能矩阵 Venera主要由四大功能模块构成，共同打造完整的漫画阅读体验： 功能模块核心价值适用场景阅读界面提供沉浸式阅读体验日常漫画阅读资源管理统一管理本地与网络漫画个人漫画收藏整理搜索发现多平台资源聚合搜索寻找新漫画资源系统配置个性化应用设置优化阅读体验 1.2 5个必须知道的技术亮点 * 跨平台支持：一次安装，多设备使用，包括手机、平板和电脑 * 漫画源扩展：通过JavaScript脚本扩展漫画来源，如同给阅读器添加新的"图书供应商" * 智能缓存系统：自动管理图片缓存，平衡阅读流畅度和存储空间 * 自定义阅读模式：

By Ne0inhk
【Linux】从版本控制到代码调试：Git 入门与 GDB 调试器学习指南

【Linux】从版本控制到代码调试：Git 入门与 GDB 调试器学习指南

前言  作为 Linux 开发的 “左膀右臂”，Git 管版本、gdb 调程序 —— 前者搞定代码的迭代与协同，后者专治程序里的各种 “疑难杂症”。这篇博客就从 Git 的核心概念、基础操作，讲到 gdb 的调试指令，把这俩工具的高频用法讲透，帮你把开发效率直接拉满。 目录 一、Git 版本控制器 1.1 什么是 Git？ 【问题】：什么是分布式？ 【问题】：什么是集中式？ 1.2 Git 核心概念  1.3 GitHub 与 Gitee 二、Git 基础操作 1. Git安装 2. 远程仓库与本地仓库联动（以

By Ne0inhk
分布式版本控制系统Git的安装和使用

分布式版本控制系统Git的安装和使用

🌈个人主页: Hygge_Code🔥热门专栏:从0开始学习Java | Linux学习| 计算机网络💫个人格言: “既然选择了远方，便不顾风雨兼程” 文章目录 * 一、关于版本控制 * 1. 什么是“版本控制”？ * 2. 版本控制系统（VCS）带来的好处🐦‍🔥 * 3. 版本控制系统分类 * 二、Git核心 * 1. 三大文件状态 * 2. 四步完成一次代码提交 * 三、Git实战：从环境配置到命令详解 * 1. 环境配置 * Git的安装操作 * 2. 本地操作命令 * 3. 远程操作命令 * 四、Git分支与标签 * 1. 分支：隔离开发 * 2. 标签：标记重要版本 * 五、版本库目录与编码规范 * 1. 版本库目录规范

By Ne0inhk

大文件上传避坑指南：vue-simple-uploader在Vue3下的进度条异常解决方案（附GitHub修改版）

大文件上传交互优化实战：从进度条异常到源码级修复的Vue3深度适配 最近在重构一个内部资源管理平台，前端技术栈从Vue2全面升级到Vue3，原本以为只是简单的版本迁移，没想到在文件上传这个看似基础的功能上遇到了不小的挑战。我们使用的是vue-simple-uploader这个在Vue2时代表现相当稳定的分片上传组件，但在Vue3环境下，上传进度条经常卡在某个百分比不动，重传、暂停按钮时灵时不灵，用户反馈体验极差。 这不仅仅是技术兼容性问题，更直接影响了用户的核心操作体验。想象一下，用户上传一个3GB的设计文件，进度条显示到78%就停滞不前，用户无法判断是正在上传还是已经失败，这种不确定性带来的焦虑感会严重影响产品口碑。经过几天的深度排查和源码分析，我找到了一套完整的解决方案，不仅解决了进度条异常问题，还对组件的交互逻辑进行了优化。 1. 问题定位：为什么Vue3环境下上传组件会“失常” 1.1 Vue3响应式系统的变化带来的连锁反应 Vue3引入的Composition API和新的响应式系统（基于Proxy）虽然性能更优，但也带来了一些兼容性挑战。vue-simple-u

By Ne0inhk