对于pyspark中的循环错误答案

【问题标题】：For Looping error in pyspark对于pyspark中的循环错误
【发布时间】：2017-05-02 19:59:08
【问题描述】：

我面临以下问题：我有一个列表，我需要将其与数据框（acc_name）中的列元素进行比较。我正在使用以下循环函数，但它只返回 1 条记录，而它应该为我提供 30 条记录。使用 Pyspark

 bs_list = 
['AC_E11','AC_E12','AC_E13','AC_E135','AC_E14','AC_E15','AC_E155','AC_E157',
 'AC_E16','AC_E163','AC_E165','AC_E17','AC_E175','AC_E180','AC_E185',   'AC_E215','AC_E22','AC_E225','AC_E23','AC_E23112','AC_E235','AC_E245','AC_E258','AC_E25','AC_E26','AC_E265','AC_E27','AC_E275','AC_E31','AC_E39','AC_E29']


    for i in bs_list:
            bs_acc1 = (acc\
                       .filter(i == acc.acc_name)
                       .select(acc.acc_name,acc.acc_description)
                      )

bs_list 元素是 acc_name 列的子集。我正在尝试创建一个新的 DF，它将具有以下 2 列 acc_name、acc_description。它将仅包含列表 bs_list 中存在的元素值的详细信息请让我知道我哪里出错了？

【问题讨论】：

您正在尝试为列表中的元素迭代 RDD。而是尝试使用 RDD 中的每个元素迭代列表。

标签： python python-2.7 pyspark ipython pyspark-sql

【解决方案1】：

那是因为，每次您在i 上进行过滤时，都会在循环中创建一个新的数据框 bs_acc1。所以它必须只显示属于 bs_list 中最后一个值的 1 行，即 'AC_E29' 的行

一种方法是与自身重复联合，因此以前的结果也保留在数据框中 -

# create a empty dataframe, give schema which is appropriate to your data below

bs_acc1 = sqlContext.createDataFrame(sc.emptyRDD(), schema)

for i in bs_list:
            bs_acc1 = bs_acc1.union(
                         acc\
                            .filter(i == acc_fil.acc_name)
                            .select(acc.acc_name,acc.acc_description)
                       )

更好的方法是根本不做循环 -

from pyspark.sql.functions import *
bs_acc1 = acc.where(acc.acc_name.isin(bs_list))

【讨论】：

我尝试了第二种解决方案，但出现以下错误：条件应为字符串或列
对不起，我的错，还有更好的方法。请参阅上面的更新答案
我尝试了两种解决方案。两者都有效。这清楚地表明我哪里出错了。谢谢

【解决方案2】：

您还可以将bs_list 转换为具有acc_name 列的数据框，然后加入acc 数据框。

bs_rdd = spark.sparkContext.parallelize(bs_list)
bs_df = bs_rdd.map(lambda x: Row(**{'acc_name':x})).toDF()
bs_join_df = bs_df.join(acc, on='acc_name')
bs_join_df.show()

【讨论】：