如何循环遍历pyspark中的每一行dataFrame答案

【问题标题】：How to loop through each row of dataFrame in pyspark如何循环遍历pyspark中的每一行dataFrame
【发布时间】：2016-07-20 20:15:41
【问题描述】：

例如

sqlContext = SQLContext(sc)

sample=sqlContext.sql("select Name ,age ,city from user")
sample.show()

上面的语句在终端打印整个表格。但我想使用for 或while 访问该表中的每一行以执行进一步的计算。

【问题讨论】：

我相信我提供了正确的答案。您可以选择或提供反馈以进行改进吗？

标签： apache-spark dataframe for-loop pyspark apache-spark-sql

【解决方案1】：

你根本做不到。 DataFrames，与其他分布式数据结构一样，不是iterable，只能使用专用的高阶函数和/或SQL方法访问。

你当然可以collect

for row in df.rdd.collect():
    do_something(row)

或转换toLocalIterator

for row in df.rdd.toLocalIterator():
    do_something(row)

并如上所示在本地进行迭代，但这超出了使用 Spark 的所有目的。

【讨论】：

新手问题：由于从数据帧中迭代已经收集的数据帧“达到目的”，我应该如何选择需要进一步处理的行？
做了一些阅读，看起来用where() 形成一个新的数据框将是正确的 Spark 方式。

【解决方案2】：

如果您想对 DataFrame 对象中的每一行执行某些操作，请使用 map。这将允许您对每一行执行进一步的计算。这相当于从0 到len(dataset)-1 循环整个数据集。

请注意，这将返回 PipelinedRDD，而不是 DataFrame。

【讨论】：

【解决方案3】：

要“循环”并利用 Spark 的并行计算框架，您可以定义自定义函数并使用 map。

def customFunction(row):

   return (row.name, row.age, row.city)

sample2 = sample.rdd.map(customFunction)

或

sample2 = sample.rdd.map(lambda x: (x.name, x.age, x.city))

然后，自定义函数将应用于数据帧的每一行。请注意，sample2 将是 RDD，而不是数据框。

如果您要执行更复杂的计算，可能需要映射。如果你只需要添加一个简单的派生列，你可以使用withColumn，返回一个数据框。

sample3 = sample.withColumn('age2', sample.age + 2)

【讨论】：

【解决方案4】：

在 python 中使用列表推导，您可以仅使用两行将一整列值收集到一个列表中：

df = sqlContext.sql("show tables in default")
tableList = [x["tableName"] for x in df.rdd.collect()]

在上面的例子中，我们返回了数据库'default'中的一个表列表，但是同样可以通过替换sql()中使用的查询来调整。

或更多缩写：

tableList = [x["tableName"] for x in sqlContext.sql("show tables in default").rdd.collect()]

对于您的三列示例，我们可以创建一个字典列表，然后在 for 循环中遍历它们。

sql_text = "select name, age, city from user"
tupleList = [{name:x["name"], age:x["age"], city:x["city"]} 
             for x in sqlContext.sql(sql_text).rdd.collect()]
for row in tupleList:
    print("{} is a {} year old from {}".format(
        row["name"],
        row["age"],
        row["city"]))

【讨论】：

【解决方案5】：

上面

tupleList = [{name:x["name"], age:x["age"], city:x["city"]}

应该是

tupleList = [{'name':x["name"], 'age':x["age"], 'city':x["city"]}

name、age 和 city 不是变量，而只是字典的键。

【讨论】：

代码行 2 右侧是否缺少方括号？
当您没有解决原始问题时，不要将其作为答案发布，而是更喜欢评论或建议对部分正确的答案进行编辑。

【解决方案6】：

Give A Try Like this

    result = spark.createDataFrame([('SpeciesId','int'), ('SpeciesName','string')],["col_name", "data_type"]); 
    for f in result.collect(): 
        print (f.col_name)

【讨论】：

【解决方案7】：

这可能不是最佳做法，但您可以简单地使用 collect() 定位特定列，将其导出为行列表，然后循环遍历列表。

假设这是你的 df：

+----------+----------+-------------------+-----------+-----------+------------------+ 
|      Date|  New_Date|      New_Timestamp|date_sub_10|date_add_10|time_diff_from_now|
+----------+----------+-------------------+-----------+-----------+------------------+ 
|2020-09-23|2020-09-23|2020-09-23 00:00:00| 2020-09-13| 2020-10-03| 51148            | 
|2020-09-24|2020-09-24|2020-09-24 00:00:00| 2020-09-14| 2020-10-04| -35252           |
|2020-01-25|2020-01-25|2020-01-25 00:00:00| 2020-01-15| 2020-02-04| 20963548         |
|2020-01-11|2020-01-11|2020-01-11 00:00:00| 2020-01-01| 2020-01-21| 22173148         |
+----------+----------+-------------------+-----------+-----------+------------------+

遍历日期列中的行：

rows = df3.select('Date').collect()

final_list = []
for i in rows:
    final_list.append(i[0])

print(final_list)

【讨论】：