教程:使用PySpark DataFrames砖

什么是DataFrame ?

DataFrame是一个二维标签数据结构可能不同类型的列。你能想到的DataFrame像电子表格,SQL表或词典系列对象。Apache火花DataFrames提供一组丰富的功能(选择列、过滤、连接、聚集),允许您来有效地解决常见的数据分析问题。

Apache火花DataFrames抽象之上的弹性分布数据集(抽样)。火花DataFrames和火花SQL使用一个统一的规划和优化引擎,允许您获得几乎相同的性能在所有受支持的语言在砖(Python、SQL、Scala和R)。

创建一个DataFrame Python

大多数Apache返回DataFrame火花查询。这包括阅读从表,从文件加载数据,操作转换数据。

您还可以创建一个火花从列表或熊猫DataFrame DataFrame,比如下面的例子:

进口熊猫作为pd数据=[[1,“伊利亚”),(2,“Teo”),(3,“方”]]pdf=pd。DataFrame(数据,列=(“id”,“名称”])df1=火花。createDataFrame(pdf)df2=火花。createDataFrame(数据,模式=“id,名称字符串”)

读表DataFrame

砖使用三角洲湖默认所有表。您可以很容易地加载表DataFrames,比如下面的例子:

火花。读。表(“< catalog_name >, < schema_name >。< table_name > ")

数据加载到DataFrame从文件

你可以从许多支持加载数据文件格式。下面的例子使用了一个可用的数据集/ databricks-datasets目录,可以从大部分工作区。看到样本数据集。

df=(火花。读。格式(“csv”)。选项(“头”,“真正的”)。选项(“inferSchema”,“真正的”)。负载(“/ databricks-datasets /样本/ population-vs-price / data_geo.csv”))

分配DataFrame转换步骤

大多数的结果返回DataFrame火花转换。你可以分配这些结果返回给一个DataFrame变量,类似于如何使用ct,临时观点,或DataFrames其他系统。

结合DataFrames和加入工会

DataFrames使用标准的SQL连接操作语义。加入返回合并后的结果两个DataFrames基于提供的匹配条件和连接类型。下面的例子是一个内连接,也就是默认值:

joined_df=df1。加入(df2,如何=“内心”,在=“id”)

您可以添加的行DataFrame到另一个使用union操作,如以下示例:

unioned_df=df1。联盟(df2)

过滤器DataFrame行

您可以过滤DataFrame使用行.filter ()或其中()。没有区别在性能或语法,见下面的例子:

filtered_df=df。过滤器(“id > 1”)filtered_df=df。在哪里(“id > 1”)

使用过滤选择一个子集的行返回DataFrame或修改。

从一个DataFrame选择列

您可以选择通过一个或多个列列名称.select (),比如下面的例子:

select_df=df。选择(“id”,“名称”)

您可以组合选择和过滤查询来限制返回的行和列。

subset_df=df。过滤器(“id > 1”)。选择(“名称”)

查看DataFrame

查看这些数据以表格格式,您可以使用砖显示()下面的示例命令,如:

显示(df)

打印数据模式

火花使用术语模式引用列的名称和数据类型DataFrame。

你可以打印使用的模式.printSchema ()方法,如以下示例:

df。printSchema()

节省DataFrame表

砖使用三角洲湖默认所有表。您可以保存的内容DataFrame表使用下面的语法:

df。写。saveAsTable(“< table_name >”)

写一个DataFrame文件的集合

大多数火花应用程序被设计成工作在大型数据集和工作在一个分布式的方式,并引发写出一个目录的文件,而不是单个文件。许多数据系统配置为读取这些文件的目录。砖推荐使用的表在filepaths对于大多数应用程序。

下面的示例JSON文件的保存目录:

df。写。格式(“json”)。保存(“/ tmp / json_data”)

在PySpark运行SQL查询

火花DataFrames提供一个选项来将SQL与Python。

的selectExpr ()方法允许您指定每一列作为一个SQL查询,比如下面的例子:

显示(df。selectExpr(“id”,“上(名称)作为big_name”))

您可以导入expr ()函数pyspark.sql.functions使用SQL语法一列将被指定的任何地方,如以下示例:

从pyspark.sql.functions进口expr显示(df。选择(“id”,expr(“低(名称)作为little_name”)))

您还可以使用spark.sql ()在Python内核运行任意SQL查询,如以下示例:

query_df=火花。sql(“从< table_name > SELECT *”)

因为逻辑是在Python执行内核和所有SQL查询是作为string传递的,您可以使用Python格式化参数化SQL查询,如以下示例:

table_name=“my_table”query_df=火花。sql(f“SELECT *{table_name}”)