PySpark 学习笔记二

谁践踏了优雅 2022-06-01 00:13 227阅读 0赞

**2 DataFrames**  
和Python的dataframe类似，pyspark也有dataframe，其处理速度远快于无结构的RDD。

Spark 2.0 用 SparkSession 代替了SQLContext。各种 Spark contexts，包括:  
HiveContext, SQLContext, StreamingContext,以及SparkContext  
全部合并到了SparkSession中，以后仅用这一个包作为读取数据的入口。

**2.1 创建 DataFrames**  
准备工作：

>>> import pyspark
    >>> from pyspark.sql import SparkSession
    >>> spark = SparkSession.builder \
            .appName("Python Spark SQL basic example") \
            .config("spark.some.config.option", "some-value") \
            .getOrCreate()
    >>> sc = spark.sparkContext

首先创建一个 stringJSONRDD的 RDD，然后将它转为一个DataFrame。

>>> stringJSONRDD = sc.parallelize(('''
    ...                     {'id':'123',
    ...                     'name':'Katie',
    ...                     'age':19,
    ...                     'eyeColor':'brown'}''',
    ...                     '''{'id','234',
    ...                     'name':'Michael',
    ...                     'age':22,
    ...                     'eyeColor':'green'}''',
    ...                     '''{'id':'345',
    ...                     'name':'Simone',
    ...                     'age':23,
    ...                     'eyeColor':'blue'}'''))

用spark.read.json(…)方法转为DataFrame：

>>> swimmersJSON = spark.read.json(stringJSONRDD)

创建一张临时table：

swimmersJSON.createOrReplaceTempView("swimmersJSON")

请注意，创建临时表是DataFrame transformation 直到执行一个action（例如，执行SQL查询）才会执行。

**2.2 DataFrame API query**  
用.show() 方法会默认打印前 10 行。

>>> swimmersJSON.show()

*  SQL 查询

>>> spark.sql("select * from swimmersJSON").collect()

有两种不同的方法把已存在的RDDs 转换为DataFrames 或者Datasets：  
一是， 用reflection指定，即用printSchema() 方法定义schema。

>>> swimmersJSON.printSchema()
    root
     |-- age: long (nullable = true)
     |-- eyeColor: string (nullable = true)
     |-- id: string (nullable = true)
     |-- name: string (nullable = true)

二是，在程序里自己定义schema。

>>> from pyspark.sql.types import *
    # Generate comma delimited data
    >>> stringCSVRDD = sc.parallelize([
    ... (123,'Katie',19,'brown'),
    ... (234,'Michael',22,'green'),
    ... (345,'Simone',23,'blue')])
    # Specify schema
    >>> schema = StructType([
    ... StructField('id',LongType(),True),
    ... StructField('name',StringType(),True),
    ... StructField("age", LongType(), True),
    ... StructField("eyeColor", StringType(), True)])

StructField类被分解为：  
• name : 该字段的名字  
• dataType : 该字段的数据类型  
• nullable : 指定此字段的值是否为空

将我们创建的schema应用于stringJSONRDD RDD（即生成的.csv数据）并创建一个临时视图，以便我们可以使用SQL查询它：

# Apply the schema to the RDD and Create DataFrame
    >>> swimmers = spark.createDataFrame(stringCSVRDD,schema)

# Creates a temporary view using the DataFrame
    >>> swimmers.createOrReplaceTempView('swimmers')
    >>> swimmers.printSchema()
    root
     |-- id: long (nullable = true)
     |-- name: string (nullable = true)
     |-- age: long (nullable = true)
     |-- eyeColor: string (nullable = true)

**2.2.1 Querying with the DataFrame API**

*  查询行数

>>> swimmers.count()
    3

*  过滤条件

# Get the id, age where age = 22
    >>> swimmers.select('id','age').filter('age=22').show()
    +---+---+
    | id|age|
    +---+---+
    |234| 22|
    +---+---+
    
    # Another way to write the above query is below
    >>> swimmers.select(swimmers.id,swimmers.age).filter(swimmers.age == 22).show()
    +---+---+
    | id|age|
    +---+---+
    |234| 22|
    +---+---+
    
    # Get the name, eyeColor where eyeColor like 'b%'
    >>> swimmers.select("name", "eyeColor").filter("eyeColor like 'b%'").show()
    +------+--------+
    |  name|eyeColor|
    +------+--------+
    | Katie|   brown|
    |Simone|    blue|
    +------+--------+

**2.2.2 Querying with the DataFrame API**  
我们在前面已经创建了视图，现在可以使用SQL语句进行查询。

*  查询行数

>>> spark.sql('select count(1) from swimmers').show()
    +--------+
    |count(1)|
    +--------+
    |       3|
    +--------+

*  过滤条件

# Get the id, age where age = 22 in SQL
    >>> spark.sql('select id,age from swimmers where age=22').show()
    +---+---+
    | id|age|
    +---+---+
    |234| 22|
    +---+---+
    
    >>> spark.sql("select name, eyeColor from swimmers where eyeColor like 'b%'").show()
    +------+--------+
    |  name|eyeColor|
    +------+--------+
    | Katie|   brown|
    |Simone|    blue|
    +------+--------+

**2.3 实例应用**

# Set File Paths
    flightPerfFilePath =
    "/databricks-datasets/flights/departuredelays.csv"
    airportsFilePath =
    "/databricks-datasets/flights/airport-codes-na.txt"
    # Obtain Airports dataset
    airports = spark.read.csv(airportsFilePath, header='true',
    inferSchema='true', sep='\t')
    airports.createOrReplaceTempView("airports")
    # Obtain Departure Delays dataset
    flightPerf = spark.read.csv(flightPerfFilePath, header='true')
    flightPerf.createOrReplaceTempView("FlightPerformance")
    # Cache the Departure Delays dataset
    flightPerf.cache()
    # Query Sum of Flight Delays by City and Origin Code
    # (for Washington State)
    spark.sql("""
    select a.City,
    f.origin,
    sum(f.delay) as Delays
    from FlightPerformance f
    join airports a
    on a.IATA = f.origin
    where a.State = 'WA'
    group by a.City, f.origin
    order by sum(f.delay) desc"""
    ).show()
    
    
    +-------+------+--------+                                                       
    |   City|origin|  Delays|
    +-------+------+--------+
    |Seattle|   SEA|159086.0|
    |Spokane|   GEG| 12404.0|
    |  Pasco|   PSC|   949.0|
    +-------+------+--------+