网站开发项目总结范文,小程序怎么申请注册费用,四川任命33名干部最新,织梦如何做英文网站目录 创建 DataFrames
生成我们自己的 JSON 数据
创建 DataFrame
创建临时表
简单的 DataFrame 查询
DataFrame API 查询
SQL 查询 创建 DataFrames
通常#xff0c;您会通过使用 SparkSession#xff08;或在 PySpark shell 中调用 spark#xff09;导入数据来创建 …目录 创建 DataFrames
生成我们自己的 JSON 数据
创建 DataFrame
创建临时表
简单的 DataFrame 查询
DataFrame API 查询
SQL 查询 创建 DataFrames
通常您会通过使用 SparkSession或在 PySpark shell 中调用 spark导入数据来创建 DataFrame。
我们将讨论如何将数据导入到本地文件系统、Hadoop 分布式文件系统HDFS或其他云存储系统例如S3 或 WASB。在本文中我们将专注于在 Spark 内直接生成您自己的 DataFrame 数据或利用 Databricks 社区版中已经可用的数据源。
首先我们将不访问文件系统而是通过生成数据来创建 DataFrame。在这种情况下我们将首先创建 stringJSONRDD RDD然后将其转换为 DataFrame。这段代码片段创建了一个包含游泳者他们的 ID、姓名、年龄和眼睛颜色的 JSON 格式的 RDD。
生成我们自己的 JSON 数据
下面我们将最初生成 stringJSONRDD RDD
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})
)
现在我们已经创建了 RDD我们将使用 SparkSession 的 read.json 方法即 spark.read.json(...)将其转换为 DataFrame。我们还将使用 .createOrReplaceTempView 方法创建一个临时表。
创建 DataFrame
以下是创建 DataFrame 的代码
swimmersJSON spark.read.json(stringJSONRDD)
创建临时表
以下是创建临时表的代码
swimmersJSON.createOrReplaceTempView(swimmersJSON)
如前文所述许多 RDD 操作是转换这些转换直到执行动作操作时才执行。例如在前面的代码片段中sc.parallelize 是一个转换当使用 spark.read.json 从 RDD 转换为 DataFrame 时执行。注意在这段代码的笔记本截图中左下角附近直到包含 spark.read.json 操作的第二个单元格Spark 作业才执行。
为了进一步强调这一点在下图的右侧窗格中我们展示了执行的 DAG 图。
在下面的截图中您可以看到 Spark 作业的 parallelize 操作来自生成 RDD stringJSONRDD 的第一个单元格而 map 和 mapPartitions 操作是创建 DataFrame 所需的操作
需要注意的是parallelize、map 和 mapPartitions 都是 RDD 转换。在 DataFrame 操作 spark.read.json在本例中中不仅有 RDD 转换还有将 RDD 转换为 DataFrame 的动作。这是一个重要的说明因为即使您正在执行 DataFrame 操作要调试您的操作您需要记住您将在 Spark UI 中理解 RDD 操作。
请注意创建临时表是一个 DataFrame 转换并且在执行 DataFrame 动作之前不会执行例如要执行的 SQL 查询。
简单的 DataFrame 查询
现在您已经创建了 swimmersJSON DataFrame我们将能够在其上运行 DataFrame API 以及 SQL 查询。让我们从一个简单的查询开始显示 DataFrame 中的所有行。
DataFrame API 查询
要使用 DataFrame API 执行此操作您可以使用 show(n) 方法该方法将前 n 行打印到控制台
# DataFrame API
swimmersJSON.show()
这将给出以下输出
SQL 查询
如果您更倾向于编写 SQL 语句您可以编写以下查询
spark.sql(select * from swimmersJSON).collect()
这将给出以下输出 我们使用了 .collect() 方法它返回所有记录作为一个行对象Row objects的列表。请注意您可以对 DataFrames 和 SQL 查询使用 collect() 或 show() 方法。只要确保如果您使用 .collect()这是针对小 DataFrame 的因为它将返回 DataFrame 中的所有行并将它们从执行器移回驱动程序。您可以改用 take(n) 或 show(n)这允许您通过指定 n 来限制返回的行数
