【spark源码系列】pyspark.sql.Row介绍和使用示例

我就是我 2023-06-20 06:11 19阅读 0赞

### 0、Spark SQL和DataFrames重要的类有： ###

*  pyspark.sql.SQLContext： DataFrame和SQL方法的主入口
 *  pyspark.sql.DataFrame： 将分布式数据集分组到指定列名的数据框中
 *  pyspark.sql.Column ：DataFrame中的列
 *  pyspark.sql.Row： DataFrame数据的行
 *  pyspark.sql.HiveContext： 访问Hive数据的主入口
 *  pyspark.sql.GroupedData： 由DataFrame.groupBy()创建的聚合方法集
 *  pyspark.sql.DataFrameNaFunctions： 处理丢失数据(空数据)的方法
 *  pyspark.sql.DataFrameStatFunctions： 统计功能的方法  
    \-pyspark.sql.functions DataFrame：可用的内置函数
 *  pyspark.sql.types： 可用的数据类型列表
 *  pyspark.sql.Window： 用于处理窗口函数

### 1、class pyspark.sql.Row:DataFrame中的一行，其中的字段可以像属性一样访问。 ###

Row可以用来通过使用命名参数来创建一个行对象，字典将按名称排序

> >>> from pyspark.sql import Row  
> >>> row = Row(name="Tom", age=11)  
> >>> row  
> Row(age=11, name='Tom')  
> >>> row\['name'\], row\['age'\]  
> ('Alice', 11)  
> >>> row\['name'\]  
> 'Tom'  
> >>> row\['age'\]  
> 11  
> >>> row.name, row.age  
> ('Tom', 11)  
> >>> row.name  
> 'Tom'  
> >>> row.age  
> 11

###  2、Row也可以用来创建另一个Row像类一样，然后它可以被用来创建Row对象 ###

> >>> Person = Row("name", "age")  
> >>> Person  
> <Row(name, age)>  
> >>> Person.name  
> Traceback (most recent call last):  
>   File "<stdin>", line 1, in <module>  
>   File "/usr/bin/hadoop/software/spark/python/pyspark/sql/types.py", line 1267, in \_\_getattr\_\_  
>     idx = self.\_\_fields\_\_.index(item)  
>   File "/usr/bin/hadoop/software/spark/python/pyspark/sql/types.py", line 1263, in \_\_getattr\_\_  
>     raise AttributeError(item)  
> AttributeError: \_\_fields\_\_  
> >>> Person("Tom", 11)  
> Row(name='Tom', age=11)  
> >>> obj\_person=Person("Tom", 11)  
> >>> obj\_person.name  
> 'Alice'  
> >>> obj\_person.age  
> 11

### 3、asDict(revursice=False):作为字典返回 ###

recursive:将嵌套的Row转换为字典（默认为False）

> >>> Row(name="Tom", age=11).asDict() == \{'name': 'Tom', 'age': 11\}  
> True  
> >>> Row(name="Tom", age=11).asDict()  
> \{'age': 11, 'name': 'Tom'\}  
> >>> row = Row(key=1, value=Row(name='a', age=2))  
> >>> row.asDict() == \{'key': 1, 'value': Row(age=2, name='a')\}  
> True  
> >>> row.asDict()  
> \{'value': Row(age=2, name='a'), 'key': 1\}  
> >>> row.asDict(True) == \{'key': 1, 'value': \{'name': 'a', 'age': 2\}\}  
> True  
> >>> row.asDict(True)  
> \{'value': \{'age': 2, 'name': 'a'\}, 'key': 1\}

### 4、spark sql返回的结果就是Row对象类型 ###

> import os
> 
> import sys
> 
> from pyspark import SparkContext, SparkConf
> 
> from pyspark.sql import HiveContext, Row
> 
> reload(sys)
> 
> sys.setdefaultencoding("utf-8")
> 
> jobName = 'req\_put\_monthly\_summary'
> 
> conf = SparkConf().setAppName(jobName).setMaster("yarn-client")
> 
> sc = SparkContext(conf=conf)
> 
> hiveContext = HiveContext(sparkContext=sc)
> 
> month\_put\_req\_pv\_uv\_list = hiveContext.sql(sqlQuery=sql\_utils.sql\_get\_month\_summary\_indices.strip()).collect()
> 
> print month\_put\_req\_pv\_uv\_list  **\# spark sql返回的结果就是Row对象类型**
> 
> '''
> 
> \[Row(month=u'201911', partnercode=u'aluc', month\_req\_pv=17358472392, month\_put\_pv=282363523, month\_req\_uv=145069532, month\_put\_uv=5740328),
> 
> Row(month=u'201911', partnercode=u'tt\_out', month\_req\_pv=13331963927, month\_put\_pv=2136909552, month\_req\_uv=340156907, month\_put\_uv=93442531)\]
> 
> '''

参考：https://www.programcreek.com/python/example/100657/pyspark.sql.Row  
           https://www.jianshu.com/p/103406e59cd3