zoukankan      html  css  js  c++  java
  • Pyspark-SQL 官方 API 的一些梳理(上)

     在 Pyspark 操纵 spark-SQL 的世界里借助 session 这个客户端来对内容进行操作和计算。里面涉及到非常多常见常用的方法,本篇文章回来梳理一下这些方法和操作。

    class pyspark.sql.SparkSession 类

    下面是一个初始化 spark session 的方法,接下来我会依次来介绍相关函数代表的意义。

    >>> spark = SparkSession.builder 
    ...     .master("local") 
    ...     .appName("Word Count") 
    ...     .config("spark.some.config.option", "some-value") 
    ...     .getOrCreate()

    SparkSession.builder:

    Builder for SparkSession 这个就是生成一个 sparksession 实例。他下面有一些支持的函数

    master: 设置 spark master 的 url 。由于我们集群使用的是 spark on yarn 的模式,所以可以用选择不设置这个参数。

    appName: 在程序运行的时候看到的应用名称。

    config: 其实其他参数也是调用 .config 设置相应的 config 参数,例如 .master 就是调用了 .config("spark.master", master)。

    enableHiveSupport: 启动对 hive 的支持。例如支持 hivesql 以及 hive udf 等。

    getOrCreate: 得到一个现成的 SparkSession ,如果没有就生成一个。

    SparkSession.catalog:

    提供一个接口来操作 create drop alter query 库或者表

    SparkSession.createDataFrame:

    可以获得从 rdd python list 和 pandas df 创建 df 的能力。下面贴一下官方的例子:

    >>> l = [('Alice', 1)]
    >>> spark.createDataFrame(l).collect()
    [Row(_1=u'Alice', _2=1)]
    >>> spark.createDataFrame(l, ['name', 'age']).collect()
    [Row(name=u'Alice', age=1)]
    
    
    >>> d = [{'name': 'Alice', 'age': 1}]
    >>> spark.createDataFrame(d).collect()
    [Row(age=1, name=u'Alice')]
    
    
    >>> rdd = sc.parallelize(l)
    >>> spark.createDataFrame(rdd).collect()
    [Row(_1=u'Alice', _2=1)]
    >>> df = spark.createDataFrame(rdd, ['name', 'age'])
    >>> df.collect()
    [Row(name=u'Alice', age=1)]
    
    
    >>> from pyspark.sql import Row
    >>> Person = Row('name', 'age')
    >>> person = rdd.map(lambda r: Person(*r))
    >>> df2 = spark.createDataFrame(person)
    >>> df2.collect()
    [Row(name=u'Alice', age=1)]
    
    
    >>> from pyspark.sql.types import *
    >>> schema = StructType([
    ...    StructField("name", StringType(), True),
    ...    StructField("age", IntegerType(), True)])
    >>> df3 = spark.createDataFrame(rdd, schema)
    >>> df3.collect()
    [Row(name=u'Alice', age=1)]
    
    
    >>> spark.createDataFrame(df.toPandas()).collect()  
    [Row(name=u'Alice', age=1)]
    >>> spark.createDataFrame(pandas.DataFrame([[1, 2]])).collect()  
    [Row(0=1, 1=2)]
    
    
    >>> spark.createDataFrame(rdd, "a: string, b: int").collect()
    [Row(a=u'Alice', b=1)]
    >>> rdd = rdd.map(lambda row: row[1])
    >>> spark.createDataFrame(rdd, "int").collect()
    [Row(value=1)]
    >>> spark.createDataFrame(rdd, "boolean").collect() 
    Traceback (most recent call last):
        ...
    Py4JJavaError: ...

    SparkSession.sql:

    使用 sql 方法返回的是 df 例如:

    >>> df.createOrReplaceTempView("table1")
    >>> df2 = spark.sql("SELECT field1 AS f1, field2 as f2 from table1")
    >>> df2.collect()
    [Row(f1=1, f2=u'row1'), Row(f1=2, f2=u'row2'), Row(f1=3, f2=u'row3')]

    SparkSession.table:

    这个可以返回指定表的 df 

    >>> df.createOrReplaceTempView("table1")
    >>> df2 = spark.table("table1")
    >>> sorted(df.collect()) == sorted(df2.collect())
    True

    由于 SQLContext 在 Spark 2.0 版本以后已经和 Session 类合并了,所以两个类一起说了。

    目前看来大量 SQLContext 的方法都被整合进了 SparkSession.catalog 方法中。

    SparkSession.catalog.cacheTable(TABLE_NAME):

    缓存相关表到内存方便使用

    SparkSession.catalog.clearCache:

    清楚所有表的缓存

    SparkSession.catalog.dropTempView():

    丢弃掉本地计算使用的临时视图,如果之前这个视图之前被 cache 过了。调用这个方法将会注销 cache。如果丢弃视图成功会返回 true 否则就返回 false

    >>> sqlContext.registerDataFrameAsTable(df, "table1")
    >>> sqlContext.dropTempTable("table1")

    DataFrame.createOrReplaceTempView(TABLE_NAME):

    使用现成的 df 创建临时视图。

    >>> df.createOrReplaceTempView("people")
    >>> df2 = df.filter(df.age > 3)
    >>> df2.createOrReplaceTempView("people")
    >>> df3 = spark.sql("select * from people")
    >>> sorted(df3.collect()) == sorted(df2.collect())
    True

    SparkSession.catalog.registerFunction:

    非常重要的一个函数用于给计算套上 udf。这个函数几经变动。首先在 1.2 版本里面在 sqlContext 里面首次实现,后来在 2.0 版本被迁移到了 SparkSession.catalog 里面实现。然后在 spark2.3.0 会迁移到 SparkSession.udf 里面实现。SparkSession.udf.register 这个函数会提供功能。由于会完全兼容之前的情况,这里我们暂时把它划分到 catalog 里。

    Parameters:    
    name – name of the UDF
    f – python function
    returnType – a pyspark.sql.types.DataType object
    >>> SparkSession.catalog.registerFunction("stringLengthString", lambda x: len(x))
    >>> SparkSession.sql("SELECT stringLengthString('test')").collect()
    [Row(stringLengthString(test)=u'4')]

    >>> from pyspark.sql.types import IntegerType >>> SparkSession.catalog.registerFunction("stringLengthInt", lambda x: len(x), IntegerType()) >>> SparkSession.sql("SELECT stringLengthInt('test')").collect() [Row(stringLengthInt(test)=4)]

    >>> from pyspark.sql.types import IntegerType >>> SparkSession.udf.register("stringLengthInt", lambda x: len(x), IntegerType()) >>> SparkSession.sql("SELECT stringLengthInt('test')").collect() [Row(stringLengthInt(test)=4)]

    SparkSession.catalog.listDatabases:

    列举目前的目前可以读取的数据库

    SparkSession.catalog.listTables:

    列举目前的目前可以读取的表

    class pyspark.sql.DataFrame(jdf, sql_ctx)
    前面说完了 SparkSession 现在这个 df 类,是我们在 python 当中操纵的对象。df 就相当于我们在 py 里面获得的一个被映射的表一样。同样 spark.sql 提供给了我们非常丰富的方法。
     
    DataFrame().alias(alias):
    重命名 df 
    >>> from pyspark.sql.functions import *
    >>> df_as1 = df.alias("df_as1")
    >>> df_as2 = df.alias("df_as2")
    >>> joined_df = df_as1.join(df_as2, col("df_as1.name") == col("df_as2.name"), 'inner')
    >>> joined_df.select("df_as1.name", "df_as2.name", "df_as2.age").collect()
    [Row(name=u'Bob', name=u'Bob', age=5), Row(name=u'Alice', name=u'Alice', age=2)]

    DataFrame().checkpoint(eager=True):

    返回这个数据集的检查点版本,检查点可以用来截断这个DataFrame的逻辑计划,这在计划可能呈指数增长的迭代算法中特别有用。它将保存到SparkContext.setCheckpointDir()设置的检查点目录中的文件中。

    DataFrame().collect():

    返回所有记录的 list 。

    >>> df.collect()
    [Row(age=2, name=u'Alice'), Row(age=5, name=u'Bob')]

    DataFrame().columns():

    以 list 的格式返回所有列名

    >>> df.columns
    ['age', 'name']

    DataFrame().count():

    返回 df 所有记录一共多少条

    >>> df.count()
    2

    DataFrame().createGlobalTempView():

    通过 df 创建一个全局的临时表。他的生命周期跟 spark 应用的生命周期相关联。如果视图中已经存在这个名字了会抛出 TempTableAlreadyExistsException 的错误。

    >>> df.createGlobalTempView("people")
    >>> df2 = spark.sql("select * from global_temp.people")
    >>> sorted(df.collect()) == sorted(df2.collect())
    True
    >>> df.createGlobalTempView("people")  
    Traceback (most recent call last):
    ...
    AnalysisException: u"Temporary table 'people' already exists;"
    >>> spark.catalog.dropGlobalTempView("people")

    这个跟之前谈到的 df 方法 createOrReplaceTempView 非常类似,createOrReplaceTempView 这个生命周期是跟随 SparkSession 这个生命周期的。

     

    DataFrame().createOrReplaceGlobalTempView():

    这个对比 createGlobalTempView 不同的点是不会抛出 TempTableAlreadyExistsException 的错误,会直接替换。

    DataFrame().createTempView():

    这个对比 createOrReplaceTempView 如果表已经存在则会抛出 TempTableAlreadyExistsException 的错误。

    DataFrame().crossJoin():

    和另外一个 df 取笛卡尔积例如

    >>> df.select("age", "name").collect()
    [Row(age=2, name=u'Alice'), Row(age=5, name=u'Bob')]
    >>> df2.select("name", "height").collect() [Row(name=u'Tom', height=80), Row(name=u'Bob', height=85)]
    >>> df.crossJoin(df2.select("height")).select("age", "name", "height").collect() [Row(age=2, name=u'Alice', height=80), Row(age=2, name=u'Alice', height=85), Row(age=5, name=u'Bob', height=80), Row(age=5, name=u'Bob', height=85)]

    DataFrame().describe():

    为数字和字符串计算统计数据,例如:

    >>> df.describe(['age']).show()
    +-------+------------------+
    |summary|               age|
    +-------+------------------+
    |  count|                 2|
    |   mean|               3.5|
    | stddev|2.1213203435596424|
    |    min|                 2|
    |    max|                 5|
    +-------+------------------+
    >>> df.describe().show()
    +-------+------------------+-----+
    |summary|               age| name|
    +-------+------------------+-----+
    |  count|                 2|    2|
    |   mean|               3.5| null|
    | stddev|2.1213203435596424| null|
    |    min|                 2|Alice|
    |    max|                 5|  Bob|
    +-------+------------------+-----+

    DataFrame().distinct():

    返回新的 df ,这个新的 df 不包含重复的数据

    >>> df.distinct().count()
    2

    DataFrame().drop():

    返回一个新的 df 丢弃掉指定的列, 如果模式不包含给定的列名,则此模式为 no-op(no operation),不删除任何列。

    >>> df.drop('age').collect()
    [Row(name=u'Alice'), Row(name=u'Bob')]
    
    >>> df.drop(df.age).collect()
    [Row(name=u'Alice'), Row(name=u'Bob')]
    
    >>> df.join(df2, df.name == df2.name, 'inner').drop(df.name).collect()
    [Row(age=5, height=85, name=u'Bob')]
    
    >>> df.join(df2, df.name == df2.name, 'inner').drop(df2.name).collect()
    [Row(age=5, name=u'Bob', height=85)]
    
    >>> df.join(df2, 'name', 'inner').drop('age', 'height').collect()
    [Row(name=u'Bob')]

    DataFrame().dropDuplicates(subset=None):

    返回一个新的 df ,这个 df 里面不再有重复的记录。可选参数可以让我们选择关心的字段进行去重。

    >>> from pyspark.sql import Row
    >>> df = sc.parallelize([ 
    ...     Row(name='Alice', age=5, height=80), 
    ...     Row(name='Alice', age=5, height=80), 
    ...     Row(name='Alice', age=10, height=80)]).toDF()
    >>> df.dropDuplicates().show() +---+------+-----+ |age|height| name| +---+------+-----+ | 5| 80|Alice| | 10| 80|Alice| +---+------+-----+
    >>> df.dropDuplicates(['name', 'height']).show() +---+------+-----+ |age|height| name| +---+------+-----+ | 5| 80|Alice| +---+------+-----+

    DataFrame().dropna(how='any'thresh=Nonesubset=None): 

    返回一个省略了 null 值的行 DataFrame.dropna() 和 DataFrameNaFunctions.drop() 互为同名函数。他的 option 参数

    how: any or all ,any 代表列中有任意数据为空就会被过滤,all 代表所有列为空才会被过滤。

    thresh: 指定被过滤的空值个数,这个值会覆盖 how 参数的指定。比如 thresh=3 如果小于3个 thresh 则会被过滤。

    subset: 指定某些列被考虑在其中。

    DataFrame().dtypes:

    返回所有列名和其类型

    >>> df.dtypes
    [('age', 'int'), ('name', 'string')]

    DataFrame().fillna(valuesubset=None):

    替换空值,和na.fill() 互为同名函数。

    value: 替换的值,可以是 int, long, float, string, or dict,如果是 dict 的话 key 应当是列值, value 应该是空值的替换值,如果是 dict 则 subset 不生效。

    subset: 指定需要忽略替换的列。

    >>> df4.na.fill(50).show()
    +---+------+-----+
    |age|height| name|
    +---+------+-----+
    | 10|    80|Alice|
    |  5|    50|  Bob|
    | 50|    50|  Tom|
    | 50|    50| null|
    +---+------+-----+
    >>> df4.na.fill({'age': 50, 'name': 'unknown'}).show() +---+------+-------+ |age|height| name| +---+------+-------+ | 10| 80| Alice| | 5| null| Bob| | 50| null| Tom| | 50| null|unknown| +---+------+-------+

    DataFrame().filter(condition):

    使用条件过滤 df records,where 和 filter 是同名函数。

    >>> df.filter(df.age > 3).collect()
    [Row(age=5, name=u'Bob')]
    
    >>> df.where(df.age == 2).collect()
    [Row(age=2, name=u'Alice')]
    
    >>> df.filter("age > 3").collect()
    [Row(age=5, name=u'Bob')]
    
    >>> df.where("age = 2").collect()
    [Row(age=2, name=u'Alice')]

    DataFrame().first:

    返回第一条 df

    DataFrame().foreach(f):

    定义一个函数,会让每个 df 都执行该函数

    >>> def f(person):
    ...     print(person.name)
    >>> df.foreach(f)

    DataFrame().foreachPartition(f):

    定义一个函数,会让每个 partitions 都执行这个函数

    >>> def f(people):
    ...     for person in people:
    ...         print(person.name)
    >>> df.foreachPartition(f)

    DataFrame().groupBy(*col):

    是 group_by 的同名函数,可以使用 agg 方法对其进行各种各样的聚合, spark sql 专门有个类为其提供了非常多的处理函数。See GroupedData for all the available aggregate functions.

    >>> df.groupBy().avg().collect()
    [Row(avg(age)=3.5)]
    >>> sorted(df.groupBy('name').agg({'age': 'mean'}).collect())
    [Row(name=u'Alice', avg(age)=2.0), Row(name=u'Bob', avg(age)=5.0)]
    >>> sorted(df.groupBy(df.name).avg().collect()) [Row(name=u'Alice', avg(age)=2.0), Row(name=u'Bob', avg(age)=5.0)]
    >>> sorted(df.groupBy(['name', df.age]).count().collect()) [Row(name=u'Alice', age=2, count=1), Row(name=u'Bob', age=5, count=1)]

    DataFrame().head(n=None):

    返回头几个 df 默认情况是 1 如果超过1 会返回一个列表。

    >>> df.head()
    Row(age=2, name=u'Alice')
    >>> df.head(1) [Row(age=2, name=u'Alice')]

    DataFrame().hint(name*parameters):

    给当前的 df 加上一些 hint 

    >>> df.join(df2.hint("broadcast"), "name").show()
    +----+---+------+
    |name|age|height|
    +----+---+------+
    | Bob|  5|    85|
    +----+---+------+

    DataFrame().join(otheron=Nonehow=None):

    使用给定的条件和其他 df 进行 join。

    other: 另外一个 df

    on: a string for the join column name, a list of column names, a join expression (Column), or a list of Columns. If on is a string or a list of strings indicating the name of the join column(s), the column(s) must exist on both sides, and this performs an equi-join.

    how: str, default inner. Must be one of: innercrossouterfullfull_outerleftleft_outerrightright_outerleft_semi, and left_anti.

    >>> df.join(df2, df.name == df2.name, 'outer').select(df.name, df2.height).collect()
    [Row(name=None, height=80), Row(name=u'Bob', height=85), Row(name=u'Alice', height=None)]
    >>> df.join(df2, 'name', 'outer').select('name', 'height').collect() [Row(name=u'Tom', height=80), Row(name=u'Bob', height=85), Row(name=u'Alice', height=None)]
    >>> cond = [df.name == df3.name, df.age == df3.age] >>> df.join(df3, cond, 'outer').select(df.name, df3.age).collect() [Row(name=u'Alice', age=2), Row(name=u'Bob', age=5)]
    >>> df.join(df2, 'name').select(df.name, df2.height).collect() [Row(name=u'Bob', height=85)]
    >>> df.join(df4, ['name', 'age']).select(df.name, df.age).collect() [Row(name=u'Bob', age=5)]

    DataFrame().limit:

    限制拿多少条

    DataFrame().na:

    返回类  DataFrameNaFunctions 用于处理空值

    DataFrame().orderBy(*cols**kwargs):

    返回一个被指定 col 排序好的 df

    >>> df.sort(df.age.desc()).collect()
    [Row(age=5, name=u'Bob'), Row(age=2, name=u'Alice')]
    
    >>> df.sort("age", ascending=False).collect()
    [Row(age=5, name=u'Bob'), Row(age=2, name=u'Alice')]
    
    >>> df.orderBy(df.age.desc()).collect()
    [Row(age=5, name=u'Bob'), Row(age=2, name=u'Alice')]
    
    >>> from pyspark.sql.functions import *
    >>> df.sort(asc("age")).collect()
    [Row(age=2, name=u'Alice'), Row(age=5, name=u'Bob')]
    
    >>> df.orderBy(desc("age"), "name").collect()
    [Row(age=5, name=u'Bob'), Row(age=2, name=u'Alice')]
    
    >>> df.orderBy(["age", "name"], ascending=[0, 1]).collect()
    [Row(age=5, name=u'Bob'), Row(age=2, name=u'Alice')]

    DataFrame().printSchema(): 

    打印出该 df 的 schema

    >>> df.printSchema()
    root
     |-- age: integer (nullable = true)
     |-- name: string (nullable = true)

    DataFrame().rdd():

    将 df 转换成 rdd 返回

    DataFrame().registerTempTable():

    将 df 注册成一个临时生命周期是 SparkSession 的周期,这个跟上面  createOrReplaceTempView 互为同名函数,就是调用该函数生成的。

    DataFrame().repartition(numPartitions*cols):

    返回一个新的 df,这个新的 df 被给定的 numPartitions 数量进行 hash 重分区。numPartitions可以是指定分区或列的目标数量的int。如果它是一个列,它将被用作第一个分区列。如果没有指定,则使用默认分区数。

    >>> df.repartition(10).rdd.getNumPartitions()
    10
    >>> data = df.union(df).repartition("age")
    >>> data.show()
    +---+-----+
    |age| name|
    +---+-----+
    |  5|  Bob|
    |  5|  Bob|
    |  2|Alice|
    |  2|Alice|
    +---+-----+
    >>> data = data.repartition(7, "age")
    >>> data.show()
    +---+-----+
    |age| name|
    +---+-----+
    |  2|Alice|
    |  5|  Bob|
    |  2|Alice|
    |  5|  Bob|
    +---+-----+
    >>> data.rdd.getNumPartitions()
    7
    >>> data = data.repartition("name", "age")
    >>> data.show()
    +---+-----+
    |age| name|
    +---+-----+
    |  5|  Bob|
    |  5|  Bob|
    |  2|Alice|
    |  2|Alice|
    +---+-----+

    DataFrame().replace(to_replacevalue=Nonesubset=None):

    返回一个 df 用参数位置2的值替换掉参数位置是1的值。DataFrame.replace() and DataFrameNaFunctions.replace()互为同名函数。

    >>> df4.na.replace(10, 20).show()
    +----+------+-----+
    | age|height| name|
    +----+------+-----+
    |  20|    80|Alice|
    |   5|  null|  Bob|
    |null|  null|  Tom|
    |null|  null| null|
    +----+------+-----+
    
    >>> df4.na.replace(['Alice', 'Bob'], ['A', 'B'], 'name').show()
    +----+------+----+
    | age|height|name|
    +----+------+----+
    |  10|    80|   A|
    |   5|  null|   B|
    |null|  null| Tom|
    |null|  null|null|
    +----+------+----+

    DataFrame().select(*cols):

    返回 select 能找到的数据

    >>> df.select('*').collect()
    [Row(age=2, name=u'Alice'), Row(age=5, name=u'Bob')]
    >>> df.select('name', 'age').collect() [Row(name=u'Alice', age=2), Row(name=u'Bob', age=5)]
    >>> df.select(df.name, (df.age + 10).alias('age')).collect() [Row(name=u'Alice', age=12), Row(name=u'Bob', age=15)]

    DataFrame().show(n=20truncate=True):

    打印前 n 行数据到控制台

    >>> df
    DataFrame[age: int, name: string]
    >>> df.show() +---+-----+ |age| name| +---+-----+ | 2|Alice| | 5| Bob| +---+-----+
    >>> df.show(truncate=3) +---+----+ |age|name| +---+----+ | 2| Ali| | 5| Bob| +---+----+

    DataFrame().sort(*cols**kwargs):

    根据给定的 cols 进行排序之后返回新的 df

    >>> df.sort(df.age.desc()).collect()
    [Row(age=5, name=u'Bob'), Row(age=2, name=u'Alice')]
    >>> df.sort("age", ascending=False).collect() [Row(age=5, name=u'Bob'), Row(age=2, name=u'Alice')]
    >>> df.orderBy(df.age.desc()).collect() [Row(age=5, name=u'Bob'), Row(age=2, name=u'Alice')]
    >>> from pyspark.sql.functions import * >>> df.sort(asc("age")).collect() [Row(age=2, name=u'Alice'), Row(age=5, name=u'Bob')]
    >>> df.orderBy(desc("age"), "name").collect() [Row(age=5, name=u'Bob'), Row(age=2, name=u'Alice')]
    >>> df.orderBy(["age", "name"], ascending=[0, 1]).collect() [Row(age=5, name=u'Bob'), Row(age=2, name=u'Alice')]

    DataFrame().sort(*cols**kwargs):

    df 查看 df 的存储级别

    >>> df.storageLevel
    StorageLevel(False, False, False, False, 1)
    >>> df.cache().storageLevel StorageLevel(True, True, False, True, 1)
    >>> df2.persist(StorageLevel.DISK_ONLY_2).storageLevel StorageLevel(True, False, False, False, 2)

    存储级别又分为

    object StorageLevel {
      val NONE = new StorageLevel(false, false, false, false)
      val DISK_ONLY = new StorageLevel(true, false, false, false)
      val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
      val MEMORY_ONLY = new StorageLevel(false, true, false, true)
      val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
      val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
      val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
      val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
      val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
      val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
      val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
      val OFF_HEAP = new StorageLevel(true, true, true, false, 1)

    DataFrame().subtract(ohter):

    取目标 df ohter 和 df 的补集。

    This is equivalent to EXCEPT in SQL.

    DataFrame().take(num)
    返回 df 前 num 条数据
    >>> df.take(2)
    [Row(age=2, name=u'Alice'), Row(age=5, name=u'Bob')]

    DataFrame().toDF(*cols)

    返回新的 df 根据 list 字段顺序

    >>> df.toDF('f1', 'f2').collect()
    [Row(f1=2, f2=u'Alice'), Row(f1=5, f2=u'Bob')]

    DataFrame().toJSON(use_unicode=True):

    >>> df.toJSON().first()
    u'{"age":2,"name":"Alice"}'

    DataFrame().toLocalIterator():

    返回一个包含所有 df row 的迭代器,迭代器将消耗与此DataFrame中最大分区相同的内存。

    DataFrame().toPandas():

    将目前的 df 转换成 pandas 的 df 

    DataFrame().union(other):

    DataFrame().unionAll(other) 这个和 union 互为同名函数。2.0 版本统一使用 union 函数了。 union 函数是 sql 函数 unionall 函数的功能。

    DataFrame().withColumn(colNamecol):

    返回一个新的 df 根据给定的 colName 和 col 本身增加一列。

    >>> df.withColumn('age2', df.age + 2).collect()
    [Row(age=2, name=u'Alice', age2=4), Row(age=5, name=u'Bob', age2=7)]

    DataFrame().withColumnRenamed(existingnew):

    返回一个新的 df 重命名现在已有的一个 col 名称。

    >>> df.withColumnRenamed('age', 'age2').collect()
    [Row(age2=2, name=u'Alice'), Row(age2=5, name=u'Bob')]

    DataFrame().write():

    非流接口向外部存储写入数据

    DataFrame().writeStream():

    流接口向外部存储写入数据

    Reference:

    https://spark.apache.org/docs/2.2.0/api/python/pyspark.sql.html  pyspark.sql module

    https://blog.csdn.net/do_yourself_go_on/article/details/74739260

  • 相关阅读:
    mysql总结
    spirngmvc整合mybatis实现CRUD
    java lesson09总结
    Java lesson08 Homework
    java Lesson08总结
    【DFS】bzoj2435 [Noi2011]道路修建
    【BFS】bzoj2252 [2010Beijing wc]矩阵距离
    【BFS】bzoj1054 [HAOI2008]移动玩具
    【搜索】bzoj3109 [cqoi2013]新数独
    【搜索】【约数个数定理】[HAOI2007]反素数ant
  • 原文地址:https://www.cnblogs.com/piperck/p/10446720.html
Copyright © 2011-2022 走看看