Spark

PySQL

文章目录

• • • PySQL
    • • 使用Spark RDD实现单词计数
      • 创建RDD
      • 操作RDD
      • 持久化RDD
      • 基于数据分区的操作
      • Spark SQL
      • • 创建DataFrame对象
        • • 1. 通过`load()`方法创建DataFrame
          • 2. 通过SparkSession方式创建DataFrame
          • DataFrame的保存
          • • 1. 通过write.***()方法保存DataFrame对象
            • 2. 通过write.format()方法保存DataFrame对象
            • DataFrame的常用操作
            • • 1. Row操作
              • 2. Column操作
              • 3. 常用属性
              • 读写���据库

                使用Spark RDD实现单词计数

                #构建SparkSession和SparkContext实例spark = SparkSession.builder.master("spark://panyx:7077")    .appName("demo").getOrCreate()sc = spark.sparkContext()# 读取数据源文件，构造一个RDD            source = "/data/spark/word.txt"sc.textFile(source)  .flatMap(lambda line: line.split(" ")) # 将每行数据按空格拆分单词  .map(lambda word:(word,1)) # 将每个单词加上计数值 1  .reduceByKey(lambda a,b:a + b) # 将所有相同的单词进行聚合相加求各单词总数  .collect()   # 返回结果给Drive程序触发RDD计算(Action)# 将计算结果保存在文件中 --ActiondataSink = "data/spark/word-result"sc.saveAsTextFile(dataSink)# 关闭SparkSessionspark.stop()

                创建RDD

                ​ 三种创建方式

                （）
                • 将现有集合并行化;

                  # 构造一个RDDrdd = sc.parallelize(range(1,21,2))

                  • 加载外部存储系统的数据集;

                    # 从文本文件创建RDDrdd = sc.textFile("hdfs://path/to/file.txt")# 从CSV文件创建RDDdf = spark.read.csv("hdfs://path/to/file.csv")  rdd = df.rdd# 从JSON文件创建RDDdf = spark.read.json("hdfs://path/to/file.json")  rdd = df.rdd# 从Parquet文件创建RDDdf = spark.read.parquet("hdfs://path/to/file.parquet")  rdd = df.rdd

                    • 在现有RDD上进行转换得到新的RDD;

                      # 通过转换操作 (transformations) 来创建新的 RDD# map()、filter()、flatMap()、distinct()等new_rdd = old_rdd.map(lambda x: x * 2)

                      操作RDD

                      ​ 两种类型的操作：Transformation和Action

                      • Transformation: 是定义如何创建RDD的延迟操作，每当在RDD上执行Transformation转换时，都会生成一个新的RDD;
                        • 常见的Transformation操作：map()，flatMap()，filter()，distinct()
                        • Action: 返回一个结果给驱动程序或将结果存入存储的操作，并开始执行计算；
                          • 常见的Action操作：

                            action操作	描述
                            reduce(func)	使用函数func对RDD中的元素进行聚合计算
                            collect()	将RDD操作的所有结果返回给驱动程序。这通常对产生足够小的RDD的操作很有用
                            count()	返回RDD中的元素数量
                            first()	返回RDD的第一个元素。它的工作原理类似于take(1)函数
                            take(n)	返回RDD的前n个元素。它首先扫描一个分区，然后使用该分区的结果来估计满足该限制所需的其他分区的数量。这个方法应该只在预期得到的数组很小的情况下使用，因为所有的数据都加载到驱动程序的内存中
                            top(n)	按照指定的隐式排序从这个RDD中取出最大的n个元素，并维护排序。这与takeOrdered相反。这个方法应该只在预期得到的数组很小的情况下使用，因为所有的数据都加载到驱动程序的内存中
                            takeSample	返回一个数组，其中包含来自RDD的元素的抽样子集
                            takeOrdered(n)	返回RDD的前n个（最小的）元素，并维护排序。这和top是相反的。这个方法应该只在预期得到的数组很小的情况下使用，因为所有的数据都加载到驱动程序的内存中
                            saveAsTextFile(path)	将RDD的元素作为文本文件(或文本文件集)写入本地文件系统、HDFS或任何其他hadoop支持的文件系统的给定目录中。Spark将对每个元素调用toString，将其转换为文件中的一行文本
                            foreach(func)	在RDD的每个元素上运行函数func
                            aggregate	类似于reduce，执行聚合运算，但它可以返回具有与输入元素数据类型不同的结果。这个函数聚合每个分区的元素，然后使用给定的combine组合函数和一个中性的“零值”，对所有分区的结果进行聚合

                            针对key/value pair RDD，PySpark专门提供了一些操作。Transformation API，包括reduceByKey()、groupByKey()、sortByKey()和join()等。Action API，包括countByKey()、collectAsMap()等。

                            reduceByKey()：使用reduce()函数合并每个key的值，其重复地应用于具有多个分区的同一组RDD数据时，它首先使用reduce()函数在各个分区本地执行合并，然后跨分区发送记录以准备最终结果。aggregateByKey()：PySpark的aggregateByKey()转换操作聚合每个key的值，使用给定的聚合函数和一个中性的“零值”，并为该key返回不同类型的值combineByKey()：使用一组自定义的聚合函数组合每个key的元素，在其内部combineByKey()操作会按分区合并元素。它也是一个宽依赖的操作，在最后阶段需要shuffle数据

                            持久化RDD

                            要缓存RDD，常用到两个函数：cache()和persist()

                            （）

                            可以使用persist()方法将一个RDD标记为持久化。之所以说“标记为持久化”，是因为出现persist()语句的地方，并不会马上计算生成RDD并把它持久化，而是要等到遇到第一个Action操作触发真正计算以后，才会把计算结果进行持久化。持久化后的RDD分区将会被保留在计算节点的内存中，被后面的Action操作重复使用。

                            基于数据分区的操作

                            函数名	函数签名	描述
                            mapPartitions	mapPartitions(f, preservesPartitioning=False)	通过对该RDD的每个分区应用函数f，返回一个新的RDD
                            mapPartitionsWithIndex	mapPartitionsWithIndex(f, preservesPartitioning=False)	通过对该RDD的每个分区应用函数，同时跟踪原始分区的索引，返回一个新的RDD
                            foreachPartition	foreachPartition(f)	将一个函数f应用到这个RDD的每个分区
                            glom	glom()	返回通过将每个分区中的所有元素合并为一个列表而创建的RDD

                            1. RDD的创建

                            使用SparkContext的parallelize方法可以创建RDD：

                            from pyspark import SparkContext    sc = SparkContext("local", "Create RDD example")  data = [1, 2, 3, 4, 5]  rdd = sc.parallelize(data)

                            还可以从外部数据源创建RDD，例如从HDFS、CSV文件、数据库等：

                            # 从HDFS上的文本文件创建RDD  rdd = sc.textFile("hdfs://path/to/file.txt")

                            1. 过滤

                            使用filter函数可以根据条件筛选RDD中的元素：

                            filtered_rdd = rdd.filter(lambda x: x % 2 == 0)  # 筛选出偶数  filtered_rdd.collect()

                            1. 去重

                            使用distinct函数可以去重：

                            distinct_rdd = rdd.distinct()  # 去重，返回唯一的元素  distinct_rdd.collect()

                            1. 排序

                            使用sortBy函数可以对RDD进行排序：

                            sorted_rdd = rdd.sortBy(lambda x: x, ascending=False)  # 按降序排序  sorted_rdd.collect()

                            1. 分组

                            使用groupBy函数可以根据指定的键对RDD进行分组：

                            grouped_rdd = rdd.groupBy(lambda x: x % 2)  # 按奇偶分组  grouped_rdd.collect()  # 输出每个键及其对应的元素组，例如[(0, [0, 2, 4]), (1, [1, 3])]

                            1. 映射 (Map)

                            from pyspark import SparkContext    sc = SparkContext("local", "Map example")  rdd = sc.parallelize([1, 2, 3, 4, 5])  mappedRDD = rdd.map(lambda x: x * 2)  # [2, 4, 6, 8, 10]  mappedRDD.collect()

                            1. 交集 (Intersection)

                            rdd1 = sc.parallelize([1, 2, 3, 4])  rdd2 = sc.parallelize([3, 4, 5, 6])  intersectionRDD = rdd1.intersection(rdd2)  # [3, 4]  intersectionRDD.collect()

                            1. 并集 (Union)

                            rdd1 = sc.parallelize([1, 2, 3])  rdd2 = sc.parallelize([3, 4, 5])  unionRDD = rdd1.union(rdd2)  # [1, 2, 3, 4, 5]  unionRDD.collect()

                            1. 差集 (Difference)

                            rdd1 = sc.parallelize([1, 2, 3, 4])  rdd2 = sc.parallelize([3, 4, 5])  differenceRDD = rdd1.subtract(rdd2)  # [1, 2]  differenceRDD.collect()

                            1. 连接 (Join)

                               假设有两个键值对RDD：rdd1 和 rdd2，我们想通过键进行连接：

                            rdd1 = sc.parallelize([("a", 1), ("b", 2), ("c", 3)])  rdd2 = sc.parallelize([("a", "X"), ("b", "Y"), ("d", "Z")])  joinedRDD = rdd1.join(rdd2)  # [("a", (1, "X")), ("b", (2, "Y")), ("c", (3, None))]  joinedRDD.collect()

                            1. 统计

                               统计RDD中的元素数量：

                            rdd = sc.parallelize([1, 2, 3, 4])  count = rdd.count()  # 输出: 4，表示有4个元素。  print(count)

                            Spark SQL

                            创建DataFrame对象

                            1. 通过load()方法创建DataFrame

                            可以通过load()方法将HDFS上的格式化文件转换为DataFrame，load()默认导入的文件格式是Parquet

                            • 使用Parquet文件

                              dfUsers=spark.read.load("file:/user/hadoop/users.parquet")# 通过show()方法展示数据dfUsers.show()

                              • 使用json文件

                                dfUsers=spark.read.format("json").load("file:/user/hadoop/users.json")# 通过show()方法展示数据dfUsers.show()

                                2. 通过SparkSession方式创建DataFrame

                                SparkSession的创建

                                from pyspark.sql import SparkSession# 链式表达spark = SparkSession.builder# 构造SparkSession对象    .appName("demo")# 设置Spark应用程序名称，不设置就随机        .config("spark.some.config.option","some-value")# 设置配置选项        .master("local[4]")# 设置连接的Spark主机master的URL，local[4]在本地使用4核运行        .getOrCreate()# 如果已存在就get,否则Create

                                • 调用createDataFrame()创建DataFrame

                                  createDataFrame(data, schema)中的data用来指定创建DataFrame对象的数据，可以是RDD、Python的列表list或Pandas的DataFrame对象；schema用来指定DataFrame的数据模式，可以是pyspark.sql.types类型指定的字段名和字段名数据类型的列表

                                  data = [("apple", "apple pie"), ("banana", "banana split"), ("cherry", "cherry cola")]  df = spark.createDataFrame(data, ["fruit", "dessert"])  

                                  • 使用range(start, end, step, numPartitions)方法创建一个列名为id的DataFrame，numPartitions是分区的意思

                                  • 使用spark.read.()方法从不同类型的文件中加载数据创建DataFrame*

                                    # 使用spark.read.***()方法从不同类型的文件中加载数据创建DataFramespark.read.csv("***.csv")spark.read.json("***.json")spark.read.parquet("***.parquet")

                                    DataFrame的保存

                                    1. 通过write.***()方法保存DataFrame对象

                                    # jsondata.write.json("file:/home/data.json")# parquetdata.write.parequet("file/home/data.parquet")# csvdata.write.csv("file/home/data.csv")

                                    2. 通过write.format()方法保存DataFrame对象

                                    # jsondata.write.format("json").save("file:/home/data.json")# parquetdata.write.format("parquet").save("file:/home/data.parquet")# csvdata.write.format("csv").save("file:/home/data.csv")

                                    先将DataFrame转化成RDD再保存到文件中

                                    data.rdd.saveAsTextFile("file:/home/hadoop/grade")

                                    DataFrame的常用操作

                                    1. Row操作

                                    1. 可使用Row类的对象创建DataFrame

                                       from pyspark.sql import functions as ffrom pyspark.sql import Rowrow1 = Row(name='Wang', spark=89, python=85)   #创建Row类的对象print(type(row1))        #查看row1的类型 row2 = Row(name='Li', spark=95, python=86)row3 = Row(name='Ding', spark=90, python=88)rdd = sc.parallelize([row1,row2,row3])  #利用Row对象创建RDDdf = rdd.toDF()    #将RDD对象转换为DataFrame,可以指定新的列名df.show()

                                    ​ 重点单词parallelize

                                    1. Row类的对象的列表直接创建DataFrame

                                       DataFrame2 = spark.createDataFrame([row1,row2,row3])

                                    2. asDict()方法将其转换为字典对象

                                       row1.asDict() # {'name': 'Wang', 'spark': 89, 'python': 85}

                                    2. Column操作

                                    1. alias()方法可对输出的列重命名

                                       df.select('name',df.spark.alias("SPARK")).show(2)  #对spark列重命名

                                    2. 对列进行排列

                                       # asc()-升序，desc()-降序df.select(df.spark,df.python).orderBy(df.spark.asc()).show()

                                    3. 改变列的数据类型

                                       # 调用astype()方法改变列的数据类型。df.select(df.spark.astype('string').alias('Spark')).collect()

                                    4. 按条件筛选

                                       • betwen

                                         # between(low,upper)df.select(df.name, df.python.between(85, 87)).show()

                                         • when

                                           # when(condition, value1).otherwise(value2)df.select(df.name,df.when(df.spark > 90,1).when(df.spark    

                                           • contains

                                             # 过滤掉所有城市列中不包含字符串”Los”的行filtered_df = df.filter(~col("City").contains("Los"))filtered_df.show()

                                             • withColumn

                                               # withColumn()方法更改列的值、转换DataFrame中已存在的列的数据类型、添加或者创建一个新的列from pyspark.sql.functions import coldf3 = df.withColumn("python",col("python")*100)   #更改列的值#使用现有列添加新列df4 = df.withColumn("PYTHON",col("python")* 10)

                                               • substr

                                                 # substr(startPos,length)方法获取从startPos索引下标开始，长度为length的子字符串result = df.withColumn("substring", df["text"].substr(1, 3))  

                                    3. 常用属性

                                    1. 输出

                                       # 1. show()    show()方法输出数据,默认输出20    show(n) 输出前n条    show(truncate=false) 全部展示# 2. collect()获取所有数据到list列表，list中每个元素是Row类型    # 3. printSchema()可查看一个DataFrame对象中有那些列，这些列是什么样的数据类型，即打印出字段名称和类型# 4. count()gradedf.count()# 输出DataFrame对象的行数    # 5.使用first()、head()、take()方法获取若干行记录（1）first()返回第一行记录 gradedf.first()（2）head()获取第一行记录，head(n)获取前n行记录 gradedf.head (2)       #获取前2行记录（3）take(n)获取前n行记录 gradedf.take(2)# 6. distinctgradedf.distinct().show() # 返回一个不包含重复记录的DataFrame对象    # 7. dropDuplicatesgradedf.dropDuplicates(["Spark"]).show()  #根据Spark字段去重

                                    2. 筛选

                                       # 1. wheregradedf.where("Class ='1' and Spark = '91'").show()   # 2. dropgradedf.drop("ID","Spark").show(3) # 去除ID,Spark字段    # 3. selectgradedf.select("Class","Name","Scala").show(3,False)    #输出筛选的数据时对列进行重命名    gradedf.select("Name","Scala").withColumnRenamed("Name","NAME").show(2)# 4. aliasgradedf.select("Name",gradedf.Spark.alias("spark")).show(3) # 对列重命名# 5. selectExpr# 可以直接对指定字段调用用户自定义函数，或者指定别名等。传入字符串类型Expr参数    gradedf.selectExpr("Name","Name as Names","upper(Name)","Scala*10").show(3)

                                    3. 排序

                                       # 1. orderBy()和sort()# 默认升序    gradedf.orderBy("Spark","Scala").show(5)gradedf.sort("Spark","Scala",ascending=False).show(5)    # 2. sortWithinPartitions# 和上面的sort()方法功能类似，区别在于sortWithinPartitions()方法返回的是按Partition排好序的DataFrame对象。

                                    4. 汇总与聚合

                                       # 1. groupBy# groupBy()按某些字段分组    （1）结合count()方法统计每组的记录数        gradedf.groupBy("Class").count().show()    （2）结合max(colNames)方法获取分组指定字段colNames的最大值    只能作用于数字型字段。        gradedf.groupBy("Class").max("Scala","Spark").show()    （4）结合sum(colNames)方法获取分组指定字段的和值只能作用于数字型字段。gradedf.groupBy("Class").sum("Spark","Scala").show()（5）结合mean(colNames)方法获取分组指定字段的平均值只能作用于数字型字段。gradedf.groupBy("Class").mean("Spark","Scala").show()# 2. agg聚合操作# agg()可以同时对多个列进行操作，生成所需要的数据。from pyspark.sql import functions as f    gradedf.agg(f.min(gradedf.Spark),f.max(gradedf.Spark)).show()

                                    5. 合并

                                       unionAll(other:Dataframe)方法用于合并两个DataFrame对象，unionAll()方法并不是按照列名和并，而是按照位置合并的，对应位置的列将合并在一起，列名不同并不影响合并，两个DataFrame对象的字段数必须相同。gradedf.select("Name","Scala","Spark").unionAll( df.select ( "name", "spark","python")).show()

                                    6. 连接

                                       # join(1)单字段连接        # 通过两个DataFrame对象的一个相同字段将两个DataFrame对象连接起来。        df1.join(df2, "name").show()    #"name"是df1和df2相同的字段        (2)多字段连接        # 通过两个DataFrame对象的多个相同字段将两个DataFrame对象连接起来。        #"name"、"Java"是df1和df2相同的字段        df1.join(df2,["name", "Java"]).show()             (3)指定join连接类型    # inner, outer, left_outer, right_outer, leftsemidf1.join(df2, ["name", "Java"], "inner").show()# to系列方法主要包括toDF、toJSON、toPandas、toLocalIterator    

                                    读写数据库

                                    连接

                                    jdbcDF = spark.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/class").option("driver","com.mysql.cj.jdbc.Driver").option("dbtable", "student").option("user", "newuser").option("password", "hadoop").load()

                                    ​