如何在 PySpark 中选择 DataFrame 中的一段行？

---

PySpark 中的 DataFrame 由一个共享的数据集合定义，该集合可用于在计算机上运行并以行和列格式对数据进行结构化。行范围定义了数据集中的一条水平线（根据条件设置多个值）。通常，范围设置最低值和最高值。在 Python 中，我们有一些内置函数，如 filter()、where() 和 collect()，用于从 PySpark 中的 DataFrame 中选择一段行。

语法

以下语法在示例中使用：

createDataFrame()

这是 Python 中的一个内置方法，它接受 schema 参数来定义 DataFrame 的模式。

filter()

filter() 是 Python 中的一个内置函数，它定义并允许根据给定条件迭代特定的行或列。

where()

这是 Python 中的一个内置函数，用于根据行或列设置特定条件，并由特定元素返回。

collect()

Pyspark collect 用于访问给定数据集中的所有元素，并在循环中使用。

show()

show() 是 Python 中的内置方法，在程序末尾使用以获取结果。

安装需求：

pip install pyspark

使用 CreateDataframe()

createDataframe 遵循 PySpark 模块，它接受两个参数 - data_name（设置每行中每列的值）和 data_Columns（通过定义所有列来设置值）。

示例

在以下示例中，我们将通过构建 SparkSession.builder 并使用 appName() 设置数据库名称来启动程序，并使用 getOrCreate() 函数创建会话并将其存储在变量 spark 中。然后在变量 customer_Data 中设置客户数据库的数据值。接下来，在变量 data_Columns 中设置所有值。现在使用 createDataFrame() 方法创建 DataFrame，该方法接受两个参数 - customer_Data 和 data_Columns，它们将以表格形式对格式进行结构化并连接到 spark，并将其存储在变量 customer_DF 中。表格形式显示行和列中的值。最后，我们使用名为 show() 的方法以及 customer_DF 来获取每列的整体数据。

# Creation of SparkSession
spark = SparkSession.builder \
   .appName('CUSTOMER') \
   .getOrCreate()    
# customer DataFrame
customer_Data = [("PREM KUMAR", 1281, "AC", 40000,4000),
   ("RATAN SINGH", 1289, "HOME THEATER", 35000, 5000),
   ("DAVID K", 1221, "NIKON CAMERA", 88000, 10000),
   ("JONATHAN REDDY", 1743, "GEYSER", 15000, 500),
   ("JASPREET BRAR", 1234, "HP LAPTOP", 78000, 3564),
   ("NEIL KAMANT", 1222, "WASHING MACHINE", 25000, 2000)
]
data_Columns = ["CUSTOMER NAME","PRODUCT ID","PRODUCT NAME",
   "Actual Price","EMI PER MONTH"]
customer_DF = spark.createDataFrame(customer_Data, data_Columns)
customer_DF.show()

输出

+--------------+----------+---------------+------------+-------------+
| CUSTOMER NAME|PRODUCT ID|   PRODUCT NAME|Actual Price|EMI PER MONTH|
+--------------+----------+---------------+------------+-------------+
|    PREM KUMAR|      1281|             AC|       40000|         4000|
|   RATAN SINGH|      1289|   HOME THEATER|       35000|         5000|
|       DAVID K|      1221|   NIKON CAMERA|       88000|        10000|
|JONATHAN REDDY|      1743|         GEYSER|       15000|          500|
| JASPREET BRAR|      1234|      HP LAPTOP|       78000|         3564|
|   NEIL KAMANT|      1222|WASHING MACHINE|       25000|         2000|
+--------------+----------+---------------+------------+-------------+

使用 Filter 方法

此方法遵循上述程序，使用它允许我们设置从 DataFrame 中选择行范围的条件。

示例

在以下示例中，我们遵循前面的示例代码，并且它允许我们使用 DataFrame（即 DF）设置两个条件，该条件接受参数值作为 Actual Price，它设置价格在 25000 到 40000 之间的行范围条件，并且它将从行中找到特定范围。最后，我们使用 show() 方法获取结果。

DF.filter((DF['Actual Price'] >= 25000) & (DF['Actual Price'] <= 40000)).show()

输出

+-------------+----------+---------------+------------+-------------+
|CUSTOMER NAME|PRODUCT ID|   PRODUCT NAME|Actual Price|EMI PER MONTH|
+-------------+----------+---------------+------------+-------------+
|   PREM KUMAR|      1281|             AC|       40000|         4000|
|  RATAN SINGH|      1289|   HOME THEATER|       35000|         5000|
|  NEIL KAMANT|      1222|WASHING MACHINE|       25000|         2000|
+-------------+----------+---------------+------------+-------------+

使用 Where() 方法

此方法遵循本文中的第一个示例，它使用 where() 方法来设置 PySpark 中 DataFrame 中的行范围。

示例

在以下示例中，我们将使用内置的 where() 方法，该方法使用 and(&) 运算符接受两个条件以获取行范围。接下来，where() 方法与 show() 方法一起使用以获取结果。

DF.where((DF['EMI PER MONTH'] >= 10000) & (DF['EMI PER MONTH'] <= 38000)).show()

输出

+-------------+----------+------------+------------+-------------+
|CUSTOMER NAME|PRODUCT ID|PRODUCT NAME|Actual Price|EMI PER MONTH|
+-------------+----------+------------+------------+-------------+
|      DAVID K|      1221|NIKON CAMERA|       88000|        10000|
+-------------+----------+------------+------------+-------------+

使用 Collect() 方法

此方法遵循第一个示例，允许它迭代到特定列以从 PySpark 中的 DataFrame 中获取行范围。

示例

在以下示例中，我们将使用 for 循环，其中变量 row 通过带有 collect() 方法的 DataFrame 进行迭代，它将迭代给定 DataFrame 中的所有元素。现在它使用 if 语句设置条件，如果 Actual price 在 30000 到 50000 之间，则返回特定范围的行。接下来，它使用接受 row 作为参数的 print() 方法返回结果。

for row in DF.collect():
   if 30000 <= row['Actual Price'] <= 50000:
      print(row)

输出

Row(CUSTOMER NAME='PREM KUMAR', PRODUCT ID=1281, PRODUCT NAME='AC', Actual Price=40000, EMI PER MONTH=4000)
Row(CUSTOMER NAME='RATAN SINGH', PRODUCT ID=1289, PRODUCT NAME='HOME THEATER', Actual Price=35000, EMI PER MONTH=5000)

结论

我们讨论了从 PySpark 中的 DataFrame 中获取行范围的各种方法。所有方法都遵循第一个示例，因为该示例设置了行和列数据的完整输入结构。PySpark 已被亚马逊、沃尔玛、途易等顶级跨国公司使用，以及更多。