当前位置： 首页 > news >正文

spark的学习-06

news 2025/7/4 5:30:42

SparkSQL读写数据的方式

1）输入Source

方式一：给定读取数据源的类型和地址

spark.read.format("json").load(path)
spark.read.format("csv").load(path)
spark.read.format("parquet").load(path)

方式二：直接调用对应数据源类型的方法

spark.read.json(path)
spark.read.csv(path)
spark.read.parquet(path)

特殊参数：option，用于指定读取时的一些配置选项

spark.read.format("csv").option("sep", "\t").load(path)jdbcDF = spark.read \.format("jdbc") \.option("url", "jdbc:postgresql:dbserver") \.option("dbtable", "schema.tablename") \.option("user", "username") \.option("password", "password") \.load()

1、普通的文件读取方式：

import osfrom pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSessionif __name__ == '__main__':os.environ['JAVA_HOME'] = 'C:/Program Files/Java/jdk1.8.0_131'# 配置Hadoop的路径，就是前面解压的那个路径os.environ['HADOOP_HOME'] = r'D:\BigDate\05-Hadoop\software\hadoop-3.3.1'# 配置base环境Python解析器的路径os.environ['PYSPARK_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'  # 配置base环境Python解析器的路径os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'# 创建spark对象spark = SparkSession.builder.master("local[2]").appName("第一次构建SparkSession").config("spark.sql.shuffle.partitions", 2).getOrCreate()deptDf = spark.read.csv("../../datas/zuoye/dept.csv").toDF("deptno","dept_name","dept_address")empDf = spark.read.format("csv").option("sep","\t").load("../../datas/zuoye/emp.tsv").toDF("empno","ename","salary","comm","deptno")deptDf.createOrReplaceTempView("dept")empDf.createOrReplaceTempView("emp")# spark.sql("""#     select * from dept# """).show()## spark.sql("""#         select * from emp#     """).show()# 需求：查询统计每个部门薪资最高的前两名员工的信息以及员工所在的部门名称。spark.sql("""with t as (select e.*,d.dept_name,dense_rank() over(partition by e.deptno order by cast(salary as int) desc) paixu from emp e join dept d on d.deptno = e.deptno) select * from t where paixu <= 2""").show()spark.stop()

2、通过jdbc读取数据库数据

注意：使用jdbd读取之前，需要先将mysql5或者8 的驱动放在

import osfrom pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSessionif __name__ == '__main__':os.environ['JAVA_HOME'] = 'C:/Program Files/Java/jdk1.8.0_131'# 配置Hadoop的路径，就是前面解压的那个路径os.environ['HADOOP_HOME'] = r'D:\BigDate\05-Hadoop\software\hadoop-3.3.1'# 配置base环境Python解析器的路径os.environ['PYSPARK_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'  # 配置base环境Python解析器的路径os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'# 得到sparkSession对象spark = SparkSession.builder.master("local[2]").appName("").config("spark.sql.shuffle.partitions", 2).getOrCreate()# 处理逻辑# 读取json 数据df1 = spark.read.format("json").load("../../datas/sql/person.json")df1.show()# 另一种写法,推荐使用这一种df2 = spark.read.json("../../datas/sql/person.json")df2.show()df3 = spark.read.csv("../../datas/dept.csv")df4 = spark.read.format("csv").load("../../datas/dept.csv")# 读取分隔符为别的分隔符的文件user_schema = StructType([StructField(name="emp_id", dataType=StringType(), nullable=False),StructField(name="emp_name", dataType=StringType(), nullable=True),StructField(name="salary", dataType=DoubleType(), nullable=True),StructField(name="comm", dataType=DoubleType(), nullable=True),StructField(name="dept_id", dataType=LongType(), nullable=True)])# 使用csv 读取了一个 \t 为分隔符的文件，读取的数据字段名很随意，所以可以自定义df5 = spark.read.format("csv").option("sep","\t").load("../../datas/emp.tsv",schema=user_schema)df5.show()# 昨天的作业是否也可以有另一个写法movie_schema = StructType([StructField(name="movie_id", dataType=LongType(), nullable=False),StructField(name="movie_name", dataType=StringType(), nullable=True),StructField(name="movie_type", dataType=StringType(), nullable=True)])movieDF = spark.read.format("csv").option("sep","::").load("../../datas/zuoye/movies.dat",schema=movie_schema)movieDF.show()spark.read.load(path="../../datas/zuoye/movies.dat",format="csv",sep="::",schema=movie_schema).show()dict = {"user":"root","password":"root"}jdbcDf = spark.read.jdbc(url="jdbc:mysql://localhost:3306/spark",table="emp",properties=dict)jdbcDf.show()# jdbc的另一种写法jdbcDf2 = spark.read.format("jdbc") \.option("driver", "com.mysql.cj.jdbc.Driver") \.option("url", "jdbc:mysql://localhost:3306/spark") \.option("dbtable", "spark.dept") \.option("user", "root") \.option("password", "root").load()jdbcDf2.show()# 读取hive表中的数据# 关闭spark.stop()

3、读取集群中hive表中的数据

import osfrom pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSessionif __name__ == '__main__':os.environ['JAVA_HOME'] = 'C:/Program Files/Java/jdk1.8.0_131'# 配置Hadoop的路径，就是前面解压的那个路径os.environ['HADOOP_HOME'] = r'D:\BigDate\05-Hadoop\software\hadoop-3.3.1'# 配置base环境Python解析器的路径os.environ['PYSPARK_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'  # 配置base环境Python解析器的路径os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'# 创建spark对象spark = SparkSession \.builder \.appName("测试spark链接") \.master("local[2]") \.config("spark.sql.warehouse.dir", 'hdfs://bigdata01:9820/user/hive/warehouse') \.config('hive.metastore.uris', 'thrift://bigdata01:9083') \.config("spark.sql.shuffle.partitions", 2) \.enableHiveSupport() \.getOrCreate()spark.sql("select * from homework.game").show()spark.stop()

2）输出Sink

方式一：给定输出数据源的类型和地址

df.write.format("json").save(path)
df.write.format("csv").save(path)
df.write.format("parquet").save(path)

方式二：直接调用对应数据源类型的方法

df.write.json(path)
df.write.csv(path)
df.write.parquet(path)

特殊参数：option，用于指定输出时的一些配置选项

df.write \
.format("jdbc") \
.option("url", "jdbc:postgresql:dbserver") \
.option("dbtable", "schema.tablename") \
.option("user", "username") \
.option("password", "password") \
.save()

append: 追加模式，当数据存在时，继续追加

overwrite: 覆写模式，当数据存在时，覆写以前数据，存储当前最新数据；

error/errorifexists: 如果目标存在就报错，默认的模式

ignore: 忽略，数据存在时不做任何操作

df.write.mode(saveMode="append").format("csv").save(path)

保存为普通格式：

import osfrom pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSessionif __name__ == '__main__':os.environ['JAVA_HOME'] = 'C:/Program Files/Java/jdk1.8.0_131'# 配置Hadoop的路径，就是前面解压的那个路径os.environ['HADOOP_HOME'] = r'D:\BigDate\05-Hadoop\software\hadoop-3.3.1'# 配置base环境Python解析器的路径os.environ['PYSPARK_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'  # 配置base环境Python解析器的路径os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'# 创建spark对象spark = SparkSession.builder.master("local[2]").appName("第一次构建SparkSession").config("spark.sql.shuffle.partitions", 2).getOrCreate()df = spark.read.json("../../datas/person.json")df.createOrReplaceTempView("person")rsDf = spark.sql("""select name,age from person where age = (select max(age) from person)""")rsDf.write.csv("hdfs://bigdata01:9820/result")spark.stop()

如果你的hdfs没有关闭安全模式的话，会报一个错误，使用下面命令关闭安全模式即可

Hdfs 关闭安全模式
hdfs dfsadmin -safemode leave

保存到本地数据库：

import osfrom pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSessionif __name__ == '__main__':os.environ['JAVA_HOME'] = 'C:/Program Files/Java/jdk1.8.0_131'# 配置Hadoop的路径，就是前面解压的那个路径os.environ['HADOOP_HOME'] = r'D:\BigDate\05-Hadoop\software\hadoop-3.3.1'# 配置base环境Python解析器的路径os.environ['PYSPARK_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'  # 配置base环境Python解析器的路径os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'# 创建spark对象spark = SparkSession.builder.master("local[2]").appName("第一次构建SparkSession").config("spark.sql.shuffle.partitions", 2).getOrCreate()df = spark.read.format("csv").option("sep","\t").load("../../datas/zuoye/emp.tsv") \.toDF("id","name","sal","comm","deptno")# 本地数据库jdbc连接df.write.format("jdbc") \.option("driver","com.mysql.cj.jdbc.Driver") \.option("url","jdbc:mysql://localhost:3306/homework") \.option("dbtable","emp1") \.option("user","root") \.option("password","123456") \.save(mode="overwrite")spark.stop()

保存到集群的hive中：

这里的spark环境是yarn，执行代码之前，需要先打开hdfs、yarn、hive（metastore和hiveserver2）

注意：如果没有使用schema，直接用toDf导入的时候，也可以导入成功，但是表中的字段类型就全都是String类型的，使用schema导入的时候可以指定字段的数据类型

import osfrom pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession
from pyspark.sql.types import StringType, StructType, StructField, IntegerTypeif __name__ == '__main__':os.environ['JAVA_HOME'] = 'C:/Program Files/Java/jdk1.8.0_131'# 配置Hadoop的路径，就是前面解压的那个路径os.environ['HADOOP_HOME'] = r'D:\BigDate\05-Hadoop\software\hadoop-3.3.1'# 配置base环境Python解析器的路径os.environ['PYSPARK_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'  # 配置base环境Python解析器的路径os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'# 创建spark对象spark = SparkSession \.builder \.appName("测试本地连接hive") \.master("local[2]") \.config("spark.sql.warehouse.dir", 'hdfs://bigdata01:9820/user/hive/warehouse') \.config('hive.metastore.uris', 'thrift://bigdata01:9083') \  #9083 metastore的端口号.config("spark.sql.shuffle.partitions", 2) \.enableHiveSupport() \.getOrCreate()# 如果没有这样指定表头的数据类型，导入hive中的数据类型全都是String 类型的# 指定了之后，表头的数据类型就是指定的了dept_schema = StructType([StructField(name="id", dataType=IntegerType(), nullable=False),StructField(name="name", dataType=StringType(), nullable=True),StructField(name="address", dataType=StringType(), nullable=True),])df = spark.read.format("csv").load("../../datas/zuoye/dept.csv",schema=dept_schema)df.show()df.write.saveAsTable("homework.dept1")spark.stop()
连接DataGrip

通过DataGrip连接 spark（读取的是hive中的数据）

/opt/installs/spark/sbin/start-thriftserver.sh --hiveconf hive.server2.thrift.port=10001 --hiveconf hive.server2.thrift.bind.host=bigdata01 --master yarn --conf spark.sql.shuffle.partitions=2

一个综合案例：

1-统计查询每个省份的总销售额【订单金额要小于1万】：省份、订单金额

2-统计查询销售额最高的前3个省份中，统计各省份单日销售额超过1000的各省份的店铺个数

  • 只对销售额最高的前3个省份做统计：将这三个省份的数据过滤出来

  • 统计每个省份每个店铺每天的销售额超过1000的店铺个数

  • 省份、店铺id【去重】、销售额、天

3-统计查询销售额最高的前3个省份中，每个省份的平均订单金额

  • 只对销售额最高的前3个省份做统计：将这三个省份的数据过滤出来

  • 按照省份分组，求订单金额平均值

4-统计查询销售额最高的前3个省份中，每个省份的每种支付类型的占比

  • 只对销售额最高的前3个省份做统计：将这三个省份的数据过滤出来

  • 支付类型：微信、刷卡、支付宝、现金

  • 支付类型的占比 = 类型支付个数 / 总个数

  • 分组：每个省份每种类型支付的个数 / 每个省份总支付个数

  • 省份、支付类型

读取数据变成DataFrame，并对不合法的数据进行清洗【过滤、转换】

  • 订单金额超过10000的订单不参与统计

  • storeProvince不为空：None， 也不为 ‘null’值

  • 只保留需要用到的字段，将字段名称转换成Python规范：a_b_c

  • 并对时间戳进行转换成日期，获取天

  • 对订单金额转换为decimal类型

import osfrom pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSessionif __name__ == '__main__':os.environ['JAVA_HOME'] = 'C:/Program Files/Java/jdk1.8.0_131'# 配置Hadoop的路径，就是前面解压的那个路径os.environ['HADOOP_HOME'] = r'D:\BigDate\05-Hadoop\software\hadoop-3.3.1'# 配置base环境Python解析器的路径os.environ['PYSPARK_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'  # 配置base环境Python解析器的路径os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe'# 创建spark对象spark = SparkSession.builder.master("local[2]").appName("第一次构建SparkSession").config("spark.sql.shuffle.partitions", 2).getOrCreate()df = spark.read.json("../../datas/function/retail.json")df.createOrReplaceTempView("orders")# spark.sql("""#     select * from orders# """).show()'''- 核心字段- "storeProvince"：订单所在的省份信息，例如：湖南省- "storeID"：订单所产生的店铺ID，例如：4064- "receivable"：订单收款金额：例如：22.5- "payType"：订单支付方式，例如：alipay- "dateTS"：订单产生时间，例如：1563758583000'''#数据清洗clearDf = spark.sql("""select storeProvince store_province,storeID store_id,cast(receivable as decimal(10,2)) receivable,payType pay_type,from_unixtime(dateTS/1000,"yyyy-MM-dd") date_tSfrom orders where receivable < 10000 and storeProvince is not null and storeProvince != 'null'""")clearDf.createOrReplaceTempView("clear_orders")# 1-统计查询每个省份的总销售额【订单金额要小于1万】：省份、订单金额df1 = spark.sql("""select store_province,sum(receivable) total_money from clear_orders group by store_province""")df1.createOrReplaceTempView("province_total_money")# 统计查询销售额最高的前3个省份中，统计各省份单日销售额超过1000的各省份的店铺个数# 只对销售额最高的前3个省份做统计：将这三个省份的数据过滤出来qiansanDf = spark.sql("""select * from clear_orders where store_province in (select store_province from province_total_money order by total_money desc limit 3)""")qiansanDf.createOrReplaceTempView("qs_province")# 统计每个省份每个店铺每天的销售额超过1000的店铺个数# spark.sql("""#     select store_province,count(distinct store_id) from qs_province group by store_province having sum(receivable) > 1000# """).show()spark.sql("""with t as(select date_ts,store_id,store_province from qs_province group by date_ts,store_id,store_province having sum(receivable) > 1000)select store_province,count(distinct store_id) store_num from t group by store_province""").show()qiansanDf.createOrReplaceTempView("qs_details")# 统计查询销售额最高的前3个省份中，统计各省份单日销售额超过1000的各省份的店铺个数spark.sql("""with t as(select date_ts,store_id,store_province from qs_details group by date_ts,store_id,store_province having sum(receivable) > 1000)select store_province,count(distinct store_id) store_num from t group by store_province""").show()# 每个省份的平均订单金额spark.sql("""select store_province,round(avg(receivable),2) avg_receivable from qs_details group by store_province""").show()# 每个省份的每种支付类型的占比spark.sql("""with t as(select store_province,pay_type,count(1) total_order from  qs_details group by store_province,pay_type)select store_province,pay_type,round(total_order/(sum(total_order) over(partition by store_province )),2) rate from t """).show()spark.stop()

http://www.lryc.cn/news/482193.html

相关文章：

  • Linux C/C++ Socket 编程
  • Flutter错误： uses-sdk:minSdkVersion 16 cannot be smaller than version 21 declared
  • Spark 的容错机制：保障数据处理的稳定性与高效性
  • TCP可靠连接的建立和释放，TCP报文段的格式，UDP简单介绍
  • LLMs之PDF：zeroX(一款PDF到Markdown 的视觉模型转换工具)的简介、安装和使用方法、案例应用之详细攻略
  • 开源数据库 - mysql - mysql-server-8.4(gtid主主同步+ keepalived热切换)部署方案
  • Java全栈体系路线
  • 【Unity基础】Unity中如何导入字体？
  • 使用NVIDIA GPU加速FFmpeg视频压制：完全指南
  • Python学习：scipy是什么？
  • spark的学习-05
  • SQL注入（SQL Injection）详解
  • 深入解析 OpenHarmony 构建系统-2-目录结构与核心组件
  • 网络安全应急响应(归纳)
  • 【网络协议栈】网络层（上）网络层的基本理解、IP协议格式、网络层分组（内附手画分析图 简单易懂）
  • 数据库类型介绍
  • 一步一步从asp.net core mvc中访问asp.net core WebApi
  • linux中kubectl命令使用
  • Linux 系统结构
  • ESP32-S3设备智能化升级，物联网无线AI语音交互，让生活更加便捷和有趣
  • Python的函数（补充浅拷贝和深拷贝）
  • oracle查询字段类型长度等字段信息
  • C语言 | Leetcode C语言题解之第559题N叉树的最大深度
  • 光流法（Optical Flow）
  • Rancher的安装
  • 【Linux】获得同一子网下当前在线设备IP/Latency/MAC 通过nmap指定CIDR扫描当前在线设备
  • Ubuntu22.04安装DataEase
  • Taro React-Native IOS 打包发布
  • 【卷积神经网络CNN】基于深度学习动物图像识别系统（完整系统源码+数据库+开发笔记+详细部署教程+启动教程）✅
  • 图像处理椒盐噪声