我试图在pyspark中运行一个for循环，该循环需要为算法过滤变量。
下面是我的Dataframedfïu prods的一个示例：

+----------+--------------------+--------------------+
|ID        |        NAME        |           TYPE     |
+----------+--------------------+--------------------+
|    7983  |SNEAKERS 01         |            Sneakers|
|    7034  |SHIRT 13            |               Shirt|
|    3360  |SHORTS 15           |               Short|

我想迭代一个id列表，从算法中获得匹配，然后过滤产品的类型。
我创建了一个函数来获取以下类型：

def get_type(ID_PROD):
    return [row[0] for row in df_prods.filter(df_prods.ID == ID_PROD).select("TYPE").collect()]

想要它回来：

print(get_type(7983))
Sneakers

但我发现两个问题：
1-这样做需要很长时间（比我在python上做类似的事情要长）
2-它返回一个字符串数组类型：['sneakers']，当我尝试过滤产品时，会发生以下情况：

type = get_type(7983)
df_prods.filter(df_prods.type == type)
java.lang.RuntimeException: Unsupported literal type class java.util.ArrayList [Sneakers]

有人知道在Pypark上更好的方法吗？
事先非常感谢。我学习Pypark很困难。