SpringBoot操作spark处理hdfs文件的操作方法

2025-01-10 04:50

本文主要是介绍SpringBoot操作spark处理hdfs文件的操作方法,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

《SpringBoot操作spark处理hdfs文件的操作方法》本文介绍了如何使用SpringBoot操作Spark处理HDFS文件,包括导入依赖、配置Spark信息、编写Controller和Ser...

SpringBoot操作spark处理hdfs文件

SpringBoot操作spark处理hdfs文件的操作方法

1、导入xgOnM依赖

<!--        spark依赖-->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>3.2.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>3.2.2</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-mllib -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-mllib_2.12</artifactId>
            <version>3.2.2</version>
        </dependency>

2、配置spark信息

建立一个配置文件,配置spark信息

import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
//将文件交于spring管理
@Configuration
public class SparkConfig {
    //使用yml中的配置
    @Value("${spark.master}")
    private String sparkMaster;
    @Value("${spark.appName}")
    private String sparkAppName;
    @Value("${hdfs.user}")
    private String hdfsUser;
    @Value("${hdfs.path}")
    private String hdfsPath;
    @Bean
    public SparkConf sparkConf() {
        SparkConf conf = new SparkConf();
        conf.setMaster(sparkMaster);
        conf.setAppName(sparkAppName);
        // 添加HDFS配置
        conf.set("fs.defaultFS", hdfsPath);
        conf.set("spark.hadoop.hdfs.user",hdfsUser);
        return conf;
    }
    @Bean
    public SparkSession sparkSession() {
        return SparkSession.builder()
                .config(sparkConf())
          php      .getOrCreate();
    }
}

3、controller和service

controller类

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import orgjavascript.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import xyz.zzj.traffic_main_code.service.SparkService;
@RestControllehttp://www.chinasem.cnr
@RequestMapping("/spark")
public class SparkController {
    @Autowired
    private SparkService sparkService;
    @GetMapping("/run")
    public String runSparkJob() {
        //读取Hadoop HDFS文件
        String filePath = "hdfs://192.168.44.128:9000/subwayData.csv";
        sparkService.executeHadoopSparkJob(filePath);
        return "Spark job executed successfully!";
    }
}

处理地铁数据的service

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.jsapache.hadoop.fs.Path;
import org.apache.spark.api.Java.JavASParkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import xyz.zzj.traffic_main_code.service.SparkReadHdfs;
import java.io.IOException;
import java.net.URI;
import static org.apache.spark.sql.functions.*;
@Service
public class SparkReadHdfsImpl implements SparkReadHdfs {
    private final SparkSession spark;
    @Value("${hdfs.user}")
    private String hdfsUser;
    @Value("${hdfs.path}")
    private String hdfsPath;
    @Autowired
    public SparkReadHdfsImpl(SparkSession spark) {
        this.spark = spark;
    }
    /**
     * 读取HDFS上的CSV文件并上传到HDFS
     * @param filePath
     */
    @Override
    public void sparkSubway(String filePath) {
        try {
            // 设置Hadoop配置
            JavaSparkContext jsc = JavaSparkContext.fromSparkContext(spark.sparkContext());
            Configuration hadoopConf = jsc.hadoopConfiguration();
            hadoopConf.set("fs.defaultFS", hdfsPath);
            hadoopConf.set("hadoop.user.name", hdfsUser);
            // 读取HDFS上的文件
            Dataset<Row> df = spark.read()
                    .option("header", "true") // 指定第一行是列名
                    .option("inferSchema", "true") // 自动推断列的数据类型
                    .csv(filePath);
            // 显示DataFrame的所有数据
//            df.show(Integer.MAX_VALUE, false);
            // 对DataFrame进行清洗和转换操作
            // 检查缺失值
            df.select("number", "people", "dateTime").na().drop().show();
            // 对数据进行类型转换
            Dataset<Row> df2 = df.select(
                    col("number").cast(DataTypes.IntegerType),
                    col("people").cast(DataTypes.IntegerType),
                    to_date(col("dateTime"), "yyyy年MM月dd日").alias("dateTime")
            );
            // 去重
            Dataset<Row> df3 = df2.dropDuplicates();
            // 数据过滤,确保people列没有负数
            Dataset<Row> df4 = df3.filter(col("people").geq(0));
//            df4.show();
            // 数据聚合,按dateTime分组,统计每天的总客流量
            Dataset<Row> df6 = df4.groupBy("dateTime").agg(sum("people").alias("total_people"));
//            df6.show();
            sparkForSubway(df6,"/time_subwayData.csv");
            //数据聚合,获取每天人数最多的地铁number
            Dataset<Row> df7 = df4.groupBy("dateTime").agg(max("people").alias("max_people"));
            sparkForSubway(df7,"/everyday_max_subwayData.csv");
            //数据聚合,计算每天的客流强度:每天总people除以632840
            Dataset<Row> df8 = df4.groupBy("dateTime").agg(sum("people").divide(632.84).alias("strength"));
            sparkForSubway(df8,"/everyday_strength_subwayData.csv");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    private static void sparkForSubway(Dataset<Row> df6, String hdfsPath) throws IOException {
        // 保存处理后的数据到HDFS
        df6.coalesce(1)
                .write().mode("overwrite")
                .option("header", "true")
                .csv("hdfs://192.168.44.128:9000/time_subwayData");
        // 创建Hadoop配置
        Configuration conf = new Configuration();
        // 获取FileSystem实例
        FileSystem fs = FileSystem.get(URI.create("hdfs://192.168.44.128:9000"), conf);
        // 定义临时目录和目标文件路径
        Path tempDir = new Path("/time_subwayData");
        FileStatus[] files = fs.listStatus(tempDir);
        // 检查目标文件是否存在,如果存在则删除
        Path targetFile1 = new Path(hdfsPath);
        if (fs.exists(targetFile1)) {
            fs.delete(targetFile1, true); // true 表示递归删除
        }
        for (FileStatus file : files) {
            if (file.isFile() && file.getPath().getName().startsWith("part-")) {
                Path targetFile = new Path(hdfsPath);
                fs.rename(file.getPath(), targetFile);
            }
        }
        // 删除临时目录
        fs.delete(tempDir, true);
    }
}

4、运行

  • 项目运行完后,打开浏览器
    • spark处理地铁数据
  • http://localhost:8686/spark/dispose
  • 观察spark和hdfs
    • http://192.168.44.128:8099/
    • http://192.168.44.128:9870/explorer.html#/

SpringBoot操作spark处理hdfs文件的操作方法

到此这篇关于SpringBoot操作spark处理hdfs文件的文章就介绍到这了,更多相关SpringBoot spark处理hdfs文件内容请搜索China编程(www.chinasem.cn)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程China编程(www.chinasem.cn)!

这篇关于SpringBoot操作spark处理hdfs文件的操作方法的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1153015

相关文章

SpringBoot 获取请求参数的常用注解及用法

《SpringBoot获取请求参数的常用注解及用法》SpringBoot通过@RequestParam、@PathVariable等注解支持从HTTP请求中获取参数,涵盖查询、路径、请求体、头、C... 目录SpringBoot 提供了多种注解来方便地从 HTTP 请求中获取参数以下是主要的注解及其用法:1

HTTP 与 SpringBoot 参数提交与接收协议方式

《HTTP与SpringBoot参数提交与接收协议方式》HTTP参数提交方式包括URL查询、表单、JSON/XML、路径变量、头部、Cookie、GraphQL、WebSocket和SSE,依据... 目录HTTP 协议支持多种参数提交方式,主要取决于请求方法(Method)和内容类型(Content-Ty

深度解析Java @Serial 注解及常见错误案例

《深度解析Java@Serial注解及常见错误案例》Java14引入@Serial注解,用于编译时校验序列化成员,替代传统方式解决运行时错误,适用于Serializable类的方法/字段,需注意签... 目录Java @Serial 注解深度解析1. 注解本质2. 核心作用(1) 主要用途(2) 适用位置3

深入浅出Spring中的@Autowired自动注入的工作原理及实践应用

《深入浅出Spring中的@Autowired自动注入的工作原理及实践应用》在Spring框架的学习旅程中,@Autowired无疑是一个高频出现却又让初学者头疼的注解,它看似简单,却蕴含着Sprin... 目录深入浅出Spring中的@Autowired:自动注入的奥秘什么是依赖注入?@Autowired

Spring 依赖注入与循环依赖总结

《Spring依赖注入与循环依赖总结》这篇文章给大家介绍Spring依赖注入与循环依赖总结篇,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录1. Spring 三级缓存解决循环依赖1. 创建UserService原始对象2. 将原始对象包装成工

Java中如何正确的停掉线程

《Java中如何正确的停掉线程》Java通过interrupt()通知线程停止而非强制,确保线程自主处理中断,避免数据损坏,线程池的shutdown()等待任务完成,shutdownNow()强制中断... 目录为什么不强制停止为什么 Java 不提供强制停止线程的能力呢?如何用interrupt停止线程s

SpringBoot请求参数传递与接收示例详解

《SpringBoot请求参数传递与接收示例详解》本文给大家介绍SpringBoot请求参数传递与接收示例详解,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋... 目录I. 基础参数传递i.查询参数(Query Parameters)ii.路径参数(Path Va

SpringBoot路径映射配置的实现步骤

《SpringBoot路径映射配置的实现步骤》本文介绍了如何在SpringBoot项目中配置路径映射,使得除static目录外的资源可被访问,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一... 目录SpringBoot路径映射补:springboot 配置虚拟路径映射 @RequestMapp

Java MCP 的鉴权深度解析

《JavaMCP的鉴权深度解析》文章介绍JavaMCP鉴权的实现方式,指出客户端可通过queryString、header或env传递鉴权信息,服务器端支持工具单独鉴权、过滤器集中鉴权及启动时鉴权... 目录一、MCP Client 侧(负责传递,比较简单)(1)常见的 mcpServers json 配置

GSON框架下将百度天气JSON数据转JavaBean

《GSON框架下将百度天气JSON数据转JavaBean》这篇文章主要为大家详细介绍了如何在GSON框架下实现将百度天气JSON数据转JavaBean,文中的示例代码讲解详细,感兴趣的小伙伴可以了解下... 目录前言一、百度天气jsON1、请求参数2、返回参数3、属性映射二、GSON属性映射实战1、类对象映