主頁(yè) > 知識(shí)庫(kù) > pyspark創(chuàng)建DataFrame的幾種方法

pyspark創(chuàng)建DataFrame的幾種方法
熱門(mén)標(biāo)簽：唐山智能外呼系統(tǒng)一般多少錢(qián) 哈爾濱ai外呼系統(tǒng)定制 海南400電話如何申請(qǐng) 激戰(zhàn)2地圖標(biāo)注 白銀外呼系統(tǒng) 騰訊外呼線路 公司電話機(jī)器人 陜西金融外呼系統(tǒng) 廣告地圖標(biāo)注app

pyspark創(chuàng)建DataFrame

為了便于操作，使用pyspark時(shí)我們通常將數(shù)據(jù)轉(zhuǎn)為DataFrame的形式來(lái)完成清洗和分析動(dòng)作。

RDD和DataFrame

在上一篇pyspark基本操作有提到RDD也是spark中的操作的分布式數(shù)據(jù)對(duì)象。

這里簡(jiǎn)單看一下RDD和DataFrame的類(lèi)型。

print(type(rdd))  # class 'pyspark.rdd.RDD'>
print(type(df))   # class 'pyspark.sql.dataframe.DataFrame'>

翻閱了一下源碼的定義，可以看到他們之間并沒(méi)有繼承關(guān)系。

class RDD(object):

    """
    A Resilient Distributed Dataset (RDD), the basic abstraction in Spark.
    Represents an immutable, partitioned collection of elements that can be
    operated on in parallel.
    """


class DataFrame(object):
    """A distributed collection of data grouped into named columns.

    A :class:`DataFrame` is equivalent to a relational table in Spark SQL,
    and can be created using various functions in :class:`SparkSession`::
 ...
    """

RDD是一種彈性分布式數(shù)據(jù)集，Spark中的基本抽象。表示一種不可變的、分區(qū)儲(chǔ)存的集合，可以進(jìn)行并行操作。
DataFrame是一種以列對(duì)數(shù)據(jù)進(jìn)行分組表達(dá)的分布式集合， DataFrame等同于Spark SQL中的關(guān)系表。相同點(diǎn)是，他們都是為了支持分布式計(jì)算而設(shè)計(jì)。

但是RDD只是元素的集合，但是DataFrame以列進(jìn)行分組，類(lèi)似于MySQL的表或pandas中的DataFrame。

實(shí)際工作中，我們用的更多的還是DataFrame。

使用二元組創(chuàng)建DataFrame

嘗試第一種情形發(fā)現(xiàn)，僅僅傳入二元組，結(jié)果是沒(méi)有列名稱(chēng)的。
于是我們嘗試第二種，同時(shí)傳入二元組和列名稱(chēng)。

a = [('Alice', 1)]
output = spark.createDataFrame(a).collect()
print(output)
# [Row(_1='Alice', _2=1)]

output = spark.createDataFrame(a, ['name', 'age']).collect()
print(output)
# [Row(name='Alice', age=1)]

這里collect()是按行展示數(shù)據(jù)表，也可以使用show()對(duì)數(shù)據(jù)表進(jìn)行展示。

spark.createDataFrame(a).show()
# +-----+---+
# |   _1| _2|
# +-----+---+
# |Alice|  1|
# +-----+---+

spark.createDataFrame(a, ['name', 'age']).show()
# +-----+---+
# | name|age|
# +-----+---+
# |Alice|  1|
# +-----+---+

使用鍵值對(duì)創(chuàng)建DataFrame

d = [{'name': 'Alice', 'age': 1}]
output = spark.createDataFrame(d).collect()
print(output)

# [Row(age=1, name='Alice')]

使用rdd創(chuàng)建DataFrame

a = [('Alice', 1)]
rdd = sc.parallelize(a)
output = spark.createDataFrame(rdd).collect()
print(output)
output = spark.createDataFrame(rdd, ["name", "age"]).collect()
print(output)

# [Row(_1='Alice', _2=1)]
# [Row(name='Alice', age=1)]

基于rdd和ROW創(chuàng)建DataFrame

from pyspark.sql import Row


a = [('Alice', 1)]
rdd = sc.parallelize(a)
Person = Row("name", "age")
person = rdd.map(lambda r: Person(*r))
output = spark.createDataFrame(person).collect()
print(output)

# [Row(name='Alice', age=1)]

基于rdd和StructType創(chuàng)建DataFrame

from pyspark.sql.types import *

a = [('Alice', 1)]
rdd = sc.parallelize(a)
schema = StructType(
    [
        StructField("name", StringType(), True),
        StructField("age", IntegerType(), True)
    ]
)
output = spark.createDataFrame(rdd, schema).collect()
print(output)

# [Row(name='Alice', age=1)]

基于pandas DataFrame創(chuàng)建pyspark DataFrame

df.toPandas()可以把pyspark DataFrame轉(zhuǎn)換為pandas DataFrame。

df = spark.createDataFrame(rdd, ['name', 'age'])
print(df)  # DataFrame[name: string, age: bigint]

print(type(df.toPandas()))  # class 'pandas.core.frame.DataFrame'>

# 傳入pandas DataFrame
output = spark.createDataFrame(df.toPandas()).collect()
print(output)

# [Row(name='Alice', age=1)]

創(chuàng)建有序的DataFrame

output = spark.range(1, 7, 2).collect()
print(output)
# [Row(id=1), Row(id=3), Row(id=5)]

output = spark.range(3).collect()
print(output)
# [Row(id=0), Row(id=1), Row(id=2)]

通過(guò)臨時(shí)表得到DataFrame

spark.registerDataFrameAsTable(df, "table1")
df2 = spark.table("table1")
b = df.collect() == df2.collect()
print(b)
# True

配置DataFrame和臨時(shí)表

創(chuàng)建DataFrame時(shí)指定列類(lèi)型

在createDataFrame中可以指定列類(lèi)型，只保留滿足數(shù)據(jù)類(lèi)型的列，如果沒(méi)有滿足的列，會(huì)拋出錯(cuò)誤。

a = [('Alice', 1)]
rdd = sc.parallelize(a)

# 指定類(lèi)型于預(yù)期數(shù)據(jù)對(duì)應(yīng)時(shí)，正常創(chuàng)建
output = spark.createDataFrame(rdd, "a: string, b: int").collect()
print(output)  # [Row(a='Alice', b=1)]
rdd = rdd.map(lambda row: row[1])
print(rdd)  # PythonRDD[7] at RDD at PythonRDD.scala:53

# 只有int類(lèi)型對(duì)應(yīng)上，過(guò)濾掉其他列。
output = spark.createDataFrame(rdd, "int").collect()
print(output)   # [Row(value=1)]

# 沒(méi)有列能對(duì)應(yīng)上，會(huì)拋出錯(cuò)誤。
output = spark.createDataFrame(rdd, "boolean").collect()
# TypeError: field value: BooleanType can not accept object 1 in type class 'int'>

注冊(cè)DataFrame為臨時(shí)表

spark.registerDataFrameAsTable(df, "table1")
spark.dropTempTable("table1")

獲取和修改配置

print(spark.getConf("spark.sql.shuffle.partitions"))  # 200
print(spark.getConf("spark.sql.shuffle.partitions", u"10"))  # 10
print(spark.setConf("spark.sql.shuffle.partitions", u"50"))  # None
print(spark.getConf("spark.sql.shuffle.partitions", u"10"))  # 50

注冊(cè)自定義函數(shù)

spark.registerFunction("stringLengthString", lambda x: len(x))
output = spark.sql("SELECT stringLengthString('test')").collect()
print(output)
# [Row(stringLengthString(test)='4')]

spark.registerFunction("stringLengthString", lambda x: len(x), IntegerType())
output = spark.sql("SELECT stringLengthString('test')").collect()
print(output)
# [Row(stringLengthString(test)=4)]

spark.udf.register("stringLengthInt", lambda x: len(x), IntegerType())
output = spark.sql("SELECT stringLengthInt('test')").collect()
print(output)
# [Row(stringLengthInt(test)=4)]

查看臨時(shí)表列表

可以查看所有臨時(shí)表名稱(chēng)和對(duì)象。

spark.registerDataFrameAsTable(df, "table1")
print(spark.tableNames())  # ['table1']
print(spark.tables())  # DataFrame[database: string, tableName: string, isTemporary: boolean]
print("table1" in spark.tableNames())  # True
print("table1" in spark.tableNames("default"))  # True

spark.registerDataFrameAsTable(df, "table1")
df2 = spark.tables()
df2.filter("tableName = 'table1'").first()
print(df2)  # DataFrame[database: string, tableName: string, isTemporary: boolean]

從其他數(shù)據(jù)源創(chuàng)建DataFrame

MySQL

前提是需要下載jar包。
Mysql-connector-java.jar

from pyspark import SparkContext
from pyspark.sql import SQLContext
import pyspark.sql.functions as F


sc = SparkContext("local", appName="mysqltest")
sqlContext = SQLContext(sc)
df = sqlContext.read.format("jdbc").options(
    url="jdbc:mysql://localhost:3306/mydata?user=rootpassword=mysql"
        "useUnicode=truecharacterEncoding=utf-8useJDBCCompliantTimezoneShift=true"
        "useLegacyDatetimeCode=falseserverTimezone=UTC ", dbtable="detail_data").load()
df.show(n=5)
sc.stop()

參考

RDD和DataFrame的區(qū)別
spark官方文檔 翻譯 之pyspark.sql.SQLContext

到此這篇關(guān)于pyspark創(chuàng)建DataFrame的幾種方法的文章就介紹到這了,更多相關(guān)pyspark創(chuàng)建DataFrame 內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:
  • pyspark 讀取csv文件創(chuàng)建DataFrame的兩種方法

標(biāo)簽：黔西 鷹潭 黑龍江 四川 常德 益陽(yáng) 惠州 上海

巨人網(wǎng)絡(luò)通訊聲明：本文標(biāo)題《pyspark創(chuàng)建DataFrame的幾種方法》，本文關(guān)鍵詞  pyspark,創(chuàng)建,DataFrame,的,幾種,；如發(fā)現(xiàn)本文內(nèi)容存在版權(quán)問(wèn)題，煩請(qǐng)?zhí)峁┫嚓P(guān)信息告之我們，我們將及時(shí)溝通與處理。本站內(nèi)容系統(tǒng)采集于網(wǎng)絡(luò)，涉及言論、版權(quán)與本站無(wú)關(guān)。
  • 相關(guān)文章
  • 下面列出與本文章《pyspark創(chuàng)建DataFrame的幾種方法》相關(guān)的同類(lèi)信息！
  • 本頁(yè)收集關(guān)于pyspark創(chuàng)建DataFrame的幾種方法的相關(guān)信息資訊供網(wǎng)民參考！
    • 推薦文章