【SparkSQL】DataFrame入门(重点:df代码操作、数据清洗API、通过JDBC读写数据库)

news2024/11/15 6:40:29

【大家好,我是爱干饭的猿,本文重点介绍DataFrame的组成、DataFrame的代码构建、DataFrame的入门操作、词频统计案例、电影数据分析、SparkSQL Shuffle 分区数目、SparkSQL 数据清洗API、DataFrame数据写出、DataFrame 通过JDBC读写数据库(MySQL示例)

后续会继续分享其他重要知识点总结,如果喜欢这篇文章,点个赞👍,关注一下吧】

上一篇文章:《【SparkSQL】基础入门(重点:SparkSQL和Hive的异同、SparkSQL数据抽象)》

3. DataFrame入门

3.1 DataFrame的组成

DataFrame是一个二维表结构, 那么表格结构就有无法
绕开的三个点:

  • 表结构描述

比如,在MySQL中的一张表:

  • 由许多行组成
  • 数据也被分成多个列
  • 表也有表结构信息(列、列名、列类型、列约束等)

基于这个前提,DataFrame的组成如下:

  • 在结构层面:
    • StructType对象描述整个DataFrame的表结构
    • StructField对象描述一个列的信息
  • 在数据层面
    • Row对象记录一行数据
    • Column对象记录一列数据并包含列的信息

在这里插入图片描述
如图, 在表结构层面,DataFrame的表结构由:
StructType描述,如下图:

在这里插入图片描述
一个StructField记录:列名、列类型、列是否运行为空
多个StructField组成一个StructType对象。
一个StructType对象可以描述一个DataFrame:有几个列、每个列的名字和类型、每个列是否为空

同时,一行数据描述为Row对象,如Row(1, 张三, 11)
一列数据描述为Column对象,Column对象包含一列数据和列的信息

3.2 DataFrame的代码构建

1. 基于RDD方式1

DataFrame对象可以从RDD转换而来,都是分布式数据集
其实就是转换一下内部存储的结构,转换为二维表结构

通过SparkSession对象的createDataFrame方法来将RDD转换为DataFrame
这里只传入列名称,类型从RDD中进行推断,是否允许为空默认为允许(True)

# coding:utf8

from pyspark.sql import SparkSession


if __name__ == '__main__':
    # 0. 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.\
        appName("test").\
        master("local[*]").\
        getOrCreate()
    sc = spark.sparkContext

    # 基于RDD转换成DataFrame
    rdd = sc.textFile("rent.txt").\
        map(lambda x: x.split(",")).\
        map(lambda x: (x[0], x[1], int(x[2])))

    # 构建DataFrame对象
    # 参数1 被转换的RDD
    # 参数2 指定列名, 通过list的形式指定, 按照顺序依次提供字符串名称即可
    df = spark.createDataFrame(rdd, schema=['address', 'area', 'price'])

    # 打印DataFrame的表结构
    df.printSchema()

    # 打印df中的数据
    # 参数1 表示 展示出多少条数据, 默认不传的话是20
    # 参数2 表示是否对列进行截断, 如果列的数据长度超过20个字符串长度, 后续的内容不显示以...代替
    # 如果给False 表示不阶段全部显示, 默认是True
    df.show(20, False)

    # 将DF对象转换成临时视图表, 可供sql语句查询
    df.createOrReplaceTempView("people")
    spark.sql("select * from people where price > 2000").show()

2. 基于RDD方式2

将RDD转换为DataFrame方式2:
通过StructType对象来定义DataFrame的“表结构”转换RDD

# coding:utf8

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, IntegerType, StringType

if __name__ == '__main__':
    # 0. 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.\
        appName("test").\
        master("local[*]").\
        getOrCreate()
    sc = spark.sparkContext

    # 基于RDD转换成DataFrame
    rdd = sc.textFile("rent.txt").\
        map(lambda x: x.split(",")).\
        map(lambda x: (x[0], x[1], int(x[2])))

    # 构建表结构的描述对象: StructType对象
    schema = StructType().\
        add("address", StringType(), nullable=True).\
        add("area", StringType(), nullable=True).\
        add("price", IntegerType(), nullable=True)

    # 基于StructType对象去构建RDD到DF的转换
    df = spark.createDataFrame(rdd, schema=schema)

    df.printSchema()
    df.show()

3. 基于RDD方式3

将RDD转换为DataFrame方式3:
使用RDD的toDF方法转换RDD

# coding:utf8

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, IntegerType, StringType

if __name__ == '__main__':
    # 0. 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.\
        appName("test").\
        master("local[*]").\
        getOrCreate()
    sc = spark.sparkContext

    # 基于RDD转换成DataFrame
    rdd = sc.textFile("rent.txt").\
        map(lambda x: x.split(",")).\
        map(lambda x: (x[0], x[1], int(x[2])))

    # 1. toDF的方式构建DataFrame
    df1 = rdd.toDF(["address", "area", "price"])
    df1.printSchema()
    df1.show()

    # 2. toDF的方式2 通过StructType来构建
    schema = StructType().\
        add("address", StringType(), nullable=True).\
        add("area", StringType(), nullable=True).\
        add("price", IntegerType(), nullable=True)

    df2 = rdd.toDF(schema=schema)
    df2.printSchema()
    df2.show()

4. 基于Pandas的DataFrame

将Pandas的DataFrame对象,转变为分布式的SparkSQL DataFrame对象

# coding:utf8
import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, IntegerType, StringType

if __name__ == '__main__':
    # 0. 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.\
        appName("test").\
        master("local[*]").\
        getOrCreate()
    sc = spark.sparkContext

    # 基于Pandas的DataFrame构建SparkSQL的DataFrame对象
    pdf = pd.DataFrame(
        {
            "id": [1, 2, 3],
            "name": ["张大仙", "王晓晓", "吕不为"],
            "age": [11, 21, 11]
        }
    )

    df = spark.createDataFrame(pdf)

    df.printSchema()
    df.show()

5. 读取外部数据-test

读取text数据源
使用format(“text”)读取文本数据
读取到的DataFrame只会有一个列,列名默认称之为:value

# coding:utf8
import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, IntegerType, StringType

if __name__ == '__main__':
    # 0. 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.\
        appName("test").\
        master("local[*]").\
        getOrCreate()
    sc = spark.sparkContext

    # 构建StructType, text数据源, 读取数据的特点是, 将一整行只作为`一个列`读取, 默认列名是value 类型是String
    schema = StructType().\
        add("data", StringType(), nullable=True)
    df = spark.read.format("text").schema(schema=schema).load("rent.txt")

    df.printSchema()
    df.show()

6. 读取外部数据-json

读取json数据源
使用format(“json”)读取json数据

# coding:utf8
import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, IntegerType, StringType

if __name__ == '__main__':
    # 0. 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.\
        appName("test").\
        master("local[*]").\
        getOrCreate()
    sc = spark.sparkContext

    # JSON类型自带有Schema信息
    df = spark.read.format("json").load("rent.json")
    df.printSchema()
    df.show()

7. 读取外部数据-csv

读取csv数据源
使用format(“csv”)读取csv数据

# coding:utf8
import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, IntegerType, StringType

if __name__ == '__main__':
    # 0. 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.\
        appName("test").\
        master("local[*]").\
        getOrCreate()
    sc = spark.sparkContext

    # 读取csv文件
    df = spark.read.format("csv").\
        option("sep", ",").\
        option("header", True).\
        option("encoding", "utf-8").\
        schema("address STRING, area STRING, price INT").\
        load("rent.csv")

    df.printSchema()
    df.show()

8. 读取外部数据-parquet

parquet: 是Spark中常用的一种列式存储文件格式,和Hive中的ORC差不多, 他俩都是列存储格式
parquet对比普通的文本文件的区别:

  • parquet 内置schema (列名\ 列类型\ 是否为空)
  • 存储是以列作为存储格式
  • 存储是序列化存储在文件中的(有压缩属性体积小)

pycharm查看parquet文件插件:Avro and Parquet Viewer

读取parquet数据源
使用format(“parquet”)读取parquet数据

# coding:utf8
import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, IntegerType, StringType

if __name__ == '__main__':
    # 0. 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.\
        appName("test").\
        master("local[*]").\
        getOrCreate()
    sc = spark.sparkContext

    # 读取parquet类型的文件
    df = spark.read.format("parquet").load("rent.parquet")

    df.printSchema()
    df.show()

3.3 DataFrame的入门操作

DataFrame支持两种风格进行编程,分别是:

  • DSL风格
  • SQL风格
  1. DSL语法风格
    DSL称之为:领域特定语言。
    其实就是指DataFrame的特有API
    DSL风格意思就是以调用API的方式来处理Data
    比如:df.where().limit()

  2. SQL语法风格
    SQL风格就是使用SQL语句处理DataFrame的数据
    比如:spark.sql(“SELECT * FROM xxx)

1. DSL风格代码演示

# coding:utf8
import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, IntegerType, StringType

if __name__ == '__main__':
    # 0. 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.\
        appName("test").\
        master("local[*]").\
        getOrCreate()
    sc = spark.sparkContext

    df = spark.read.format("csv").\
        schema("address STRING, area STRING, price INT").\
        load("rent.txt")

    # Column对象的获取
    address_column = df['address']
    price_column = df['price']

    # TODO: DLS风格演示
    df.select(["address", "price"]).show()
    df.select("address", "price").show()
    df.select(address_column, price_column).show()

    # 1. filter API
    df.filter("price > 2500").show()
    df.filter(df['price'] > 2500).show()

    # 2. where API
    df.where("price > 2500").show()
    df.where(df['price']> 2500).show()

    # 3. group By API
    df.groupBy("address").count().show()
    df.groupBy(df['address']).count().show()


    # df.groupBy API的返回值 GroupedData
    # GroupedData对象 不是DataFrame
    # 它是一个 有分组关系的数据结构, 有一些API供我们对分组做聚合
    # SQL: group by 后接上聚合: sum avg count min man
    # GroupedData 类似于SQL分组后的数据结构, 同样有上述5种聚合方法
    # GroupedData 调用聚合方法后, 返回值依旧是DataFrame
    # GroupedData 只是一个中转的对象, 最终还是要获得DataFrame的结果
    r = df.groupBy("address")

2. SQL风格代码演示

# coding:utf8
import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, IntegerType, StringType

if __name__ == '__main__':
    # 0. 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.\
        appName("test").\
        master("local[*]").\
        getOrCreate()
    sc = spark.sparkContext

    df = spark.read.format("csv").\
        schema("address STRING, area STRING, price INT").\
        load("rent.txt")

    # 注册成临时表
    df.createTempView("rent")  # 注册临时视图(表)
    df.createOrReplaceTempView("rent_2")  # 注册 或者 替换  临时视图
    df.createGlobalTempView("rent_3")  # 注册全局临时视图 全局临时视图在使用的时候 需要在前面带上global_temp. 前缀

    # 可以通过SparkSession对象的sql api来完成sql语句的执行
    spark.sql("SELECT area, COUNT(*) AS cnt FROM rent GROUP BY area").show()
    spark.sql("SELECT area, COUNT(*) AS cnt FROM rent_2 GROUP BY area").show()
    spark.sql("SELECT area, COUNT(*) AS cnt FROM global_temp.rent_3 GROUP BY area").show()

3.4 词频统计案例

我们来完成一个单词计数需求,使用DSL和SQL两种风格来实现。

# coding:utf8
import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, IntegerType, StringType
from pyspark.sql import functions as F

if __name__ == '__main__':
    # 0. 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.\
        appName("test").\
        master("local[*]").\
        getOrCreate()
    sc = spark.sparkContext

    # words.txt:
    # hello hadoop
    # hello spark
    # hello flink

    # TODO 1: SQL 风格进行处理
    rdd = sc.textFile("words.txt").\
        flatMap(lambda x: x.split(" ")).\
        map(lambda x: [x])

    df = rdd.toDF(["word"])

    # 注册DF为表格
    df.createTempView("words")
    spark.sql("select word, count(*) as cnt from words group by word order by cnt desc").show()

    # TODO 2: DSL 风格处理
    df = spark.read.format("text").load("words.txt")

    # withColumn方法
    # 方法功能: 对已存在的列进行操作, 返回一个新的列, 如果名字和老列相同, 那么替换, 否则作为新列存在
    df2 = df.withColumn("value", F.explode(F.split(df["value"], " ")))

    df2.groupBy("value").\
        count().\
        withColumnRenamed("value", "word").\
        withColumnRenamed("count", "cnt").\
        orderBy("cnt", ascending=False).\
        show()

3.5 电影数据分析

MovieLens数据集
MovieLens数据集包含多个用户对多部电影的评级数据,也包括电影元数据信息和用户属性信息。
下载地址: http://files.grouplens.org/datasets/movielens
介绍:下面以ml-10Ok数据集为例进行。
介绍:下载u.data文件。
u.data -由943个用户对1682个电影的10000条评分组成。每个用户至少评分20部电影。用户和电影从1号开始连续编号。数据是随机排序的。
在这里插入图片描述

需求:

  1. 查询用户平均分
  2. 查询电影平均分
  3. 查询大于平均分的电影的数量
  4. 查询高分电影中(>3)打分次数最多的用户,并求出此人打的平均分
  5. 查询每个用户的平均打分,最低打分,最高打分
  6. 查询被评分超过100次的电影,的平均分排名TOP10
# coding:utf8
import time

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StringType, IntegerType
import pandas as pd
from pyspark.sql import functions as F


if __name__ == '__main__':
    # 0. 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.\
        appName("test").\
        master("local[*]").\
        config("spark.sql.shuffle.partitions", 2).\
        getOrCreate()
    sc = spark.sparkContext

    """
    spark.sql.shuffle.partitions 参数指的是, 在sql计算中, shuffle算子阶段默认的分区数是200个.
    对于集群模式来说, 200个默认也算比较合适
    如果在local下运行, 200个很多, 在调度上会带来额外的损耗
    所以在local下建议修改比较低 比如2\4\10均可
    这个参数和Spark RDD中设置并行度的参数 是相互独立的.
    """

    # 1. 读取数据集
    schema = StructType().\
        add("user_id", StringType(), nullable=True).\
        add("movie_id", IntegerType(), nullable=True).\
        add("rank", IntegerType(), nullable=True).\
        add("ts", StringType(), nullable=True)
    df = spark.read.format("csv").\
        option("sep", "\t").\
        option("header", False).\
        schema(schema=schema).\
        load("u.data")

    df.printSchema()

    # TODO 1: 用户平均分
    df.groupBy("user_id").\
        avg("rank").\
        withColumnRenamed("avg(rank)", "avg_rank").\
        withColumn("avg_rank", F.round("avg_rank", 2)).\
        orderBy("avg_rank", ascending=False).\
        show()

    df.createTempView("movie")
    spark.sql("select user_id, round(avg(rank), 2) as avg_rank from movie group by user_id").show()

    # TODO 2: 电影的平均分查询
    df.groupBy("movie_id").\
        avg("rank").\
        withColumnRenamed("avg(rank)", "avg_rank").\
        withColumn("avg_rank", F.round("avg_rank", 2)).\
        orderBy("avg_rank", ascending=False).\
        show()

    spark.sql("select movie_id, round(avg(rank), 2) as avg_rank from movie group by movie_id order by avg_rank desc").show()

    # TODO 3: 查询大于平均分的电影的数量 # Row
    print("大于平均分电影的数量: {}".format(df.where(df['rank'] > df.select(F.avg(df['rank'])).first()['avg(rank)']).count()))

    spark.sql("select count(*) as cnt from movie where rank > (select avg(rank) from movie)").show()

    # TODO 4: 查询高分电影中(>3)打分次数最多的用户, 此人打分的平均分
    user_id = df.where("rank > 3").\
        groupBy("user_id").\
        count().\
        withColumnRenamed("count", "cnt").\
        orderBy("cnt", ascending=False).\
        limit(1).\
        first()["user_id"]
    print("user_id: {}".format(user_id))

    df.filter(df["user_id"] == user_id).\
        select(F.round(F.avg("rank"), 2)).show()

    spark.sql("select movie.user_id, round(avg(rank), 2) as avg_rank from movie, (select user_id from movie where rank > 3 group by user_id order by count(*) desc limit 1) as u where u.user_id == movie.user_id group by movie.user_id").show()

    # TODO 5: 查询每个用户的平局打分, 最低打分, 最高打分
    df.groupBy("user_id").\
        agg(
        F.round(F.avg("rank"), 2).alias("avg_rank"),
        F.min("rank").alias("min_rank"),
        F.max("rank").alias("max_rank")
    ).orderBy("avg_rank", ascending=False).\
    show()

    spark.sql("select user_id, round(avg(rank), 2) as avg_rank, min(rank) as min_rank, max(rank) as max_rank from movie group by user_id order by avg_rank desc limit 10").show()

    # TODO 6: 查询评分超过100次的电影, 的平均分 排名 TOP10
    df.groupBy("movie_id").\
        agg(
        F.count("movie_id").alias("cnt"),
        F.round(F.avg("rank"), 2).alias("avg_rank")
    ).where("cnt > 100").\
        orderBy("avg_rank", ascending=False).\
        limit(10).\
        show()

    spark.sql("select movie_id, count(rank) as cnt, round(avg(rank), 2) as avg_rank from movie group by movie_id having count(rank) > 100 order by avg_rank desc limit 10").show()

    # time.sleep(10000) 可以打开spark运行地址查看task任务执行情况 127.0.0.1:4040

"""
1. agg: 它是GroupedData对象的API, 作用是 在里面可以写多个聚合
2. alias: 它是Column对象的API, 可以针对一个列 进行改名
3. withColumnRenamed: 它是DataFrame的API, 可以对DF中的列进行改名, 一次改一个列, 改多个列 可以链式调用
4. orderBy: DataFrame的API, 进行排序, 参数1是被排序的列, 参数2是 升序(True) 或 降序 False
5. first: DataFrame的API, 取出DF的第一行数据, 返回值结果是Row对象.
# Row对象 就是一个数组, 你可以通过row['列名'] 来取出当前行中, 某一列的具体数值. 返回值不再是DF 或者GroupedData 或者Column而是具体的值(字符串, 数字等)
"""

3.6 SparkSQL Shuffle 分区数目

在这里插入图片描述

3.7 SparkSQL 数据清洗API

前面我们处理的数据实际上都是已经被处理好的规整数据,但是在大数据整个生产过程中,需要先对数据进行数据清洗,将杂乱无章的数据整理为符合后面处理要求的规整数据。

# coding:utf8

from pyspark.sql import SparkSession


if __name__ == '__main__':
    # 0. 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.\
        appName("test").\
        master("local[*]").\
        config("spark.sql.shuffle.partitions", 2).\
        getOrCreate()
    sc = spark.sparkContext

    """读取数据"""
    df = spark.read.format("csv").\
        option("sep", ";").\
        option("header", True).\
        load("../data/input/sql/people.csv")

    # TODO 1: 数据清洗: 数据去重
    # dropDuplicates 是DataFrame的API, 可以完成数据去重
    # 无参数使用, 对全部的列 联合起来进行比较, 去除重复值, 只保留一条
    df.dropDuplicates().show()

    df.dropDuplicates(['age', 'job']).show()

    # TODO 2: 数据清洗: 缺失值处理
    # dropna api是可以对缺失值的数据进行删除
    # 无参数使用, 只要列中有null 就删除这一行数据
    df.dropna().show()
    # thresh = 3表示, 最少满足3个有效列,  不满足 就删除当前行数据
    df.dropna(thresh=3).show()

    df.dropna(thresh=2, subset=['name', 'age']).show()

    # TODO 3: 缺失值处理也可以完成对缺失值进行填充
    # DataFrame的 fillna 对缺失的列进行填充
    df.fillna("loss").show()

    # 指定列进行填充
    df.fillna("N/A", subset=['job']).show()

    # 设定一个字典, 对所有的列 提供填充规则
    df.fillna({"name": "未知姓名", "age": 1, "job": "worker"}).show()

3.8 DataFrame数据写出

SparkSQL统一API写出DataFrame数据

# coding:utf8
import time

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StringType, IntegerType
import pandas as pd
from pyspark.sql import functions as F


if __name__ == '__main__':
    # 0. 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.\
        appName("test").\
        master("local[*]").\
        config("spark.sql.shuffle.partitions", 2).\
        getOrCreate()
    sc = spark.sparkContext

    # 1. 读取数据集
    schema = StructType().add("user_id", StringType(), nullable=True). \
        add("movie_id", IntegerType(), nullable=True). \
        add("rank", IntegerType(), nullable=True). \
        add("ts", StringType(), nullable=True)
    df = spark.read.format("csv"). \
        option("sep", "\t"). \
        option("header", False). \
        option("encoding", "utf-8"). \
        schema(schema=schema). \
        load("u.data")

    # 2. Write text 写出, 只能写出一个列的数据, 需要将df转换为单列df
    df.select(F.concat_ws("---", "user_id", "movie_id", "rank", "ts")).\
        write.\
        mode("overwrite").\
        format("text").\
        save("./sql/text")

    # 3. Write csv
    df.write.mode("overwrite").\
        format("csv").\
        option("sep", ";").\
        option("header", True).\
        save("./sql/csv")

    # 4. Write json
    df.write.mode("overwrite").\
        format("json").\
        save("./sql/json")

    # 5. Write parquet
    df.write.mode("overwrite").\
        format("parquet").\
        save("./sql/parquet")

3.9 DataFrame 通过JDBC读写数据库(MySQL示例)

读取JDBC 是需要有驱动的,我们读取的是 jdbc:mysql://

这个协议,也就是读取的是mysql的数据,既然如此,就需要有mysql的驱动jar包给spark程序用.
如果不给驱动jar包,会提示:No suitable Driver
在这里插入图片描述

  • 对于windows系统(使用本地解释器)(以Anaconda环境演示)
    将jar包放在: Anaconda3的安装路径下\envs\虚拟环境\Liblsite-packages\pyspark\jars

  • 对于Linux系统(使用远程解释器执行)(以Anaconda环境演示)
    将jar包放在:Anaconda3的安装路径下/envs/虚拟环境/lib/python3.8/site-packages/pyspark/jars

# coding:utf8
import time

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StringType, IntegerType
import pandas as pd
from pyspark.sql import functions as F


if __name__ == '__main__':
    # 0. 构建执行环境入口对象SparkSession
    spark = SparkSession.builder.\
        appName("test").\
        master("local[*]").\
        config("spark.sql.shuffle.partitions", 2).\
        getOrCreate()
    sc = spark.sparkContext

    # 1. 读取数据集
    schema = StructType().add("user_id", StringType(), nullable=True). \
        add("movie_id", IntegerType(), nullable=True). \
        add("rank", IntegerType(), nullable=True). \
        add("ts", StringType(), nullable=True)
    df = spark.read.format("csv"). \
        option("sep", "\t"). \
        option("header", False). \
        option("encoding", "utf-8"). \
        schema(schema=schema). \
        load("u.data")

    # TODO 1: 写出df到mysql数据库中
    df.write.mode("overwrite").\
        format("jdbc").\
        option("url", "jdbc:mysql://127.0.0.1:3306/python_learn?useSSL=false&useUnicode=true").\
        option("dbtable", "movie_date").\
        option("user", "root").\
        option("password", "123456").\
        save()

    # TODO 2: 读取MySQL数据
    df2 = spark.read.\
        format("jdbc").\
        option("url", "jdbc:mysql://127.0.0.1:3306/python_learn?useSSL=false&useUnicode=true").\
        option("dbtable", "movie_date").\
        option("user", "root").\
        option("password", "123456").\
        load()
    df2.printSchema()
    df2.show()

"""
JDBC写出, 会自动创建表的.
因为DataFrame中有表结构信息, StructType记录的 各个字段的 名称 类型  和是否运行为空
"""

3.10 总结

  1. DataFrame 在结构层面上由StructField组成列描述,由StructType构造表描述。在数据层面上,Column对象记录列数据,Row对象记录行数据
  2. DataFrame可以从RDD转换、Pandas DF转换、读取文件、读取JDBC等方法构建
  3. spark.read.format()和df.write.format() 是DataFrame读取和写出的统一化标准API
  4. SparkSQL默认在Shuffle阶段200个分区,可以修改参数获得最好性能
  5. dropDuplicates可以去重、dropna可以删除缺失值、fillna可以填充缺失值
  6. SparkSQL支持JDBC读写,可用标准API对数据库进行读写操作

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/1273392.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

linux 内核等待队列

等待队列在Linux内核中用来阻塞或唤醒一个进程,也可以用来同步对系统资源的访问,还可以实现延迟功能 在软件开发中任务经常由于某种条件没有得到满足而不得不进入睡眠状态,然后等待条件得到满足的时候再继续运行,进入运行状态。这…

虚拟数字人有什么用?有哪些应用场景?

​​过去三年,元宇宙概念进入到大众视野,虚拟数字人备受关注。抖音达人柳夜熙、洛天依、网红虚拟偶像AYAYI等,随着元宇宙的流行,数字人也逐渐成为一种趋势。据行业预测,到2030年,中国的数字人总市场规模将达…

全汉电源SN生产日期解读

新买了一个全汉的电脑电源,SN:WZ3191900030,看了几次没想明白,最后估计SN是2023年19周这样来记录日期的。问了一下京东全汉客服,果然就是这样的。那大家如果在闲鱼上看到全汉电源,就知道它的生产日期了。

Excel导入操作

<template><el-dialogwidth"500px"title"员工导入":visible"showExcelDialog"close"$emit(update:showExcelDialog, false)"><el-row type"flex" justify"center"><div class"upload-e…

Csharp(C#)无标题栏窗体拖动代码

C#&#xff08;C Sharp&#xff09;是一种现代、通用的编程语言&#xff0c;由微软公司在2000年推出。C#是一种对象导向的编程语言&#xff0c;它兼具C语言的高效性和Visual Basic语言的易学性。C#主要应用于Windows桌面应用程序、Windows服务、Web应用程序、游戏开发等领域。C…

使用vscode的remotessh插件远程连接的时候被要求重复输入密码

问题描述&#xff1a; 需要远程连接服务器&#xff0c;使用ssh&#xff0c;我用到的是vscode里面的remotessh插件。配置好config以后 HostHostNameUserPortIdentifyFile进入到了vscode的密码登录界面&#xff0c;但是一直被要求循环输入密码&#xff0c;很奇怪&#xff0c;去…

遭到美国做空机构“灰熊”做空后,人工智能公司商汤科技股价暴跌

来源&#xff1a;猛兽财经 作者&#xff1a;猛兽财经 猛兽财经获悉&#xff0c;在遭到美国做空机构Grizzly Research&#xff08;灰熊&#xff09;指控夸大收入后&#xff0c;商汤科技的股价在周二一度下跌了9.7%。 Grizzly Research在周二发布的一份报告中称&#xff0c;商汤…

分享5款靠谱好用,无广告不流氓的好软件

​ 话不多说&#xff0c;直入正题&#xff0c;全都是靠谱好用&#xff0c;无广告不流氓的好软件&#xff0c;可以先点赞收藏&#xff0c;以后慢慢用。 1.动态壁纸软件——Lively Wallpaper ​ Lively Wallpaper是一款可以将视频、GIF、网页、游戏等内容作为桌面壁纸的软件&am…

对抗产品团队中的认知偏误:给产品经理的专家建议

今天的产品经理面临着独特的挑战。他们不仅需要设计和构建创新功能&#xff0c;还必须了解这些功能将如何为客户带来价值并推进关键业务目标。如果不加以控制&#xff0c;认知偏差可能会导致您构建的内容与客户想要的内容或业务需求之间不一致。本文将详细阐述产品经理可以避免…

融合抖音生态:抖音核销工具小程序开发全指南

为了更好地与抖音生态融合&#xff0c;许多开发者开始关注抖音核销工具小程序的开发。本文将为您提供一份详尽的指南&#xff0c;帮助您了解如何开发一个完善的抖音核销工具小程序。 第一步&#xff1a;理解抖音生态 抖音提供了一系列开发者工具和API&#xff0c;包括用户授权…

上市公司数字化转型及同群效应数据集合(四种测算方法)

数据简介&#xff1a;当今世界处于高速发展的信息时代中&#xff0c;数字革命的产生催生出大量数字技术和数字信息。在数字经济时代&#xff0c;数字化转型赋予了企业新的发展动能&#xff0c;数字化转型已经成为诸多企业高质量发展的重要路径。是否需要进行数字化转型、能否及…

Serilog .net下的新兴的日志框架

Serilog .net下的新兴的日志框架 1.Serilog简介 Serilog 是针对 .NET 应用程序的流行日志记录框架。它以其灵活性、易用性和可扩展性而闻名。借助 Serilog&#xff0c;开发人员可以轻松记录应用程序中的事件、错误和消息。它支持结构化日志记录&#xff0c;能够以结构化格式存…

SpringBoot整合validation数据校验

1. 首先引入依赖 <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-validation</artifactId></dependency> 点标识进去可以发现是通过Hibernate Validator使用 Java Bean Validation 2. 属性上…

中伟视界:AI盒子中的报警预录像功能能解决什么问题?实现原理是怎样的?

现代社会智能安防已成为各行各业的重要一环&#xff0c;而AI盒子中的报警预录像功能更是智能安防的一大利器。这一功能能够解决很多安防方面的难题&#xff0c;其实现原理更是技术创新的体现。 首先&#xff0c;让我们来看看AI盒子中的报警预录像功能能解决哪些问题。在传统的安…

基于PHP的校园兼职系统的设计与开发

基于PHP的校园兼职系统的设计与开发 摘要&#xff1a;从古代至今&#xff0c;教育都是国家培养人才的手段&#xff0c;在古代教育往往都是课堂式教育&#xff0c;在课堂内老师教导学生学习&#xff0c;而随着时间的推移&#xff0c;越来越多的在校大学生已经不满足于只在课堂上…

dapper+mysql查询报Error parsing column 0 (Id=<null>)

之前的分页接口都是正常的&#xff0c;突然就报错了Error parsing column 0 (Id<null>) {"error": {"code": null,"message": "Error parsing column 0 (Id<null>)","details": "DataException: Error pa…

vue.js ——Vuex

基本概念 vue进行开发过程中有没有遇到这样一种场景&#xff0c;就是有些时候一些数据是一种通用的共享数据&#xff08;比如登录信息&#xff09;&#xff0c;那么这类数据在各个组件模块中可能都会用到&#xff0c;如果每个组件中都去后台重新获取那么势必会造成性能浪费&am…

R语言30分钟入门

1. 环境&安装 R是支持win、linux合macos的 完整参考&#xff1a;https://zhuanlan.zhihu.com/p/596324321?utm_id0 主要是安装&#xff1a;1、R环境&#xff1b;2、rstudio开发环境&#xff08;后面主要是用rstudio&#xff0c;也可以用vscode&#xff09; 1.1. rstud…

异常 Exception 02

异常 Exception 02 六、异常处理1、基本介绍2、异常处理的方式3、示意图 try-catchthrows1、介绍2、注意事项 自定义异常1、基本概念2、自定义异常的步骤3、实例4、throw和throws的区别 六、异常处理 1、基本介绍 异常处理就是当异常发生时,对异常处理的方式。 2、异常处理的…

佳易王物流快运物流单打印登记查询系统软件操作教程

一、前言&#xff08;编程应用实例系列&#xff09;&#xff1a; 佳易王物流快运物流单打印登记查询系统软件操作教程 软件有试用版&#xff0c;可以下载试用&#xff0c;了解软件操作和软件功能。 软件试用版下载可以点击最下方官网卡片 软件为绿色免安装版&#xff0c;下载…