数据输入与输出入门

概述

在数据分析项目中，最常见的任务是读取数据并以新形式将其写回，无论是最终格式还是原始数据的更新版本。

数据输入与输出流程

数据输入与输出的流程如下：

• 读取数据到 pandas 和 PySpark DataFrame 中。
• 使用 withColumnRenamed 方法 重命名 DataFrame 的列。
• 选择仅与我们分析相关的部分列。
• 将数据写回到磁盘，作为跨多个文件的分布式数据集，并通过分区和排序来提升性能。

读取数据取决于提供的数据源。在将数据读入 PySpark DataFrame 之前，可能需要进行一些预处理。输入数据可能是 CSV、JSON 或 Parquet 格式。本课程所使用的数据集为 JSON 格式，因此我们将重点放在读取 JSON 数据上。

我们还可以使用特定数据库的 JDBC 或 ODBC 驱动程序，从数据库读取数据到 PySpark DataFrame。

要运行 PySpark，首先需要初始化其会话。让我们快速了解一下如何操作。

创建 PySpark 会话

我们将编写代码来初始化环境。

创建 PySpark 会话的代码

下面的代码用于初始化环境。在代码中，我们使用 create_spark_session 函数创建一个包含四个从节点和一个主节点的 spark 执行器。这些节点共使用五个线程来完成任何 PySpark 任务。

说明

• 第 1–2 行：
  导入所需的库 SparkSession，用于创建 PySpark 会话。

• 第 3–12 行：
  定义一个用于创建 PySpark 会话的函数 create_spark_session。

  • 第 3 行： 定义该函数。
  • 第 5 行： 加载环境配置。
  • 第 6–12 行： 返回一个 PySpark 会话。
    • 第 9 行： 为会话指定名称 "SparkApp"。
    • 第 10 行： 在本地机器上创建五个线程作为逻辑核心。

• 第 13 行：
  调用该函数以创建 PySpark 会话。

• 第 14 行：
  打印提示信息，表示会话已启动。

代码成功执行后，终端将显示“Code Executed Successfully”消息。