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

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

入门篇--Python篇-3-为什么做AI开发首选Python,而不是Java或C++?

入门篇--Python篇-3-为什么做AI开发首选Python,而不是Java或C++?

作者:Weisian AI探索者 · 软件工程师 · 写给每一个想入行AI的你 哈喽,各位技术小伙伴~ 最近后台好多刚入门AI的同学问我:“想做AI开发,先学Python、Java还是C++呢?” 其实答案很明确——90%的AI开发场景,Python都是首选。 但不是说Java和C++不好,而是这两种语言的优势场景,和AI开发的核心需求不太匹配。今天就从大家实际开发会遇到的问题出发,把这个问题讲透:为什么Python能成为AI开发的“万能钥匙”,而Java、C++更适合做“专项补充”? 一、先搞懂:AI开发的核心需求是什么? 不管是做算法研究、模型训练,还是落地一个简单的AI应用(比如图像识别、聊天机器人),核心需求就3个: 1. 快速迭代:AI是“试错型”开发,比如调模型参数、改算法逻辑,可能要试几十上百次,开发效率直接决定进度; 2. 少造轮子:AI涉及大量复杂计算(矩阵运算、

私募排排网JS逆向实战:解密加密数据与Python调用全解析

1. 从抓包到逆向:一次完整的实战开场 大家好,我是老张,在数据采集和逆向这块摸爬滚打十来年了。今天想和大家分享一个非常有意思的实战案例——如何搞定私募排排网的数据加密。这个案例很典型,它不像一些简单的网站直接返回明文JSON,而是把核心数据用JS加密后返回,你需要先解密才能看到真容。整个过程就像侦探破案,一步步追踪线索,最终找到解密钥匙。如果你对爬虫、JS逆向或者金融数据采集感兴趣,这篇文章会带你走一遍完整的流程,从发现加密到最终用Python调通,我会把踩过的坑和关键技巧都讲清楚。 咱们的目标很明确,就是拿到私募排排网基金排行列表页的数据。当你打开浏览器开发者工具,查看网络请求时,会发现一个关键的XHR请求,地址是 https://sppwapi.simuwang.com/sun/ranking/fundRankV3。这个请求返回的数据乍一看是一堆乱码,或者是一个结构奇怪的JSON对象,里面有个 data 字段,内容像是经过Base64编码的密文。这就是典型的服务端加密,前端通过一段JavaScript逻辑进行解密后渲染到页面上。我们的任务就是找到这段解密逻辑,并用Python

OKX API v5新特性:python-okx库全面支持与使用指南

OKX API v5新特性:python-okx库全面支持与使用指南 【免费下载链接】python-okx 项目地址: https://gitcode.com/GitHub_Trending/py/python-okx 你还在为交易接口的复杂集成而烦恼?是否需要一个既能处理现货交易又能管理衍生品合约的Python工具?本文将带你一文掌握OKX API v5的全部新特性,通过python-okx库实现从账户管理到算法交易的全流程操作,让你的量化策略开发效率提升300%。 读完本文你将获得: * 5分钟快速上手的API接入指南 * 现货/衍生品交易的完整代码模板 * WebSocket实时行情推送的最佳实践 * 10+常见交易场景的解决方案 为什么选择python-okx库? OKX API v5作为行业领先的交易接口,提供了毫秒级的订单执行速度和全品类的交易支持。而python-okx库作为其官方推荐的Python封装,相比其他第三方库具有三大优势: 特性python-okx普通第三方库接口覆盖率100% REST API + WebSocket约60%核心接

如何安装 `.whl` 文件(Python Wheel 包)

如何安装 `.whl` 文件(Python Wheel 包)

目录标题 * 如何安装 `.whl` 文件(Python Wheel 包) * 🛠 安装前提 * 💻 安装方法(3种) * 方法 1:直接使用 pip 安装(推荐) * 方法 2:先进入文件目录再安装 * 方法 3:使用绝对路径(适合脚本中调用) * ⚠️ 常见问题解决 * 问题 1:版本不兼容错误 * 问题 2:缺少依赖 * 问题 3:权限不足 * 🔍 验证安装 * 💡 进阶技巧 如何安装 .whl 文件(Python Wheel 包) .whl 文件是 Python 的 二进制分发格式(Wheel 格式),用于快速安装 Python