我处理一个有两列的数据帧，mvv 和 count。
+---+-----+
|mvv|count|
+---+-----+
| 1 | 5 |
| 2 | 9 |
| 3 | 3 |
| 4 | 1 |
我想获得两个包含 MVV 值和计数值的列表。类似的东西
mvv = [1,2,3,4]
count = [5,9,3,1]
因此，我尝试了以下代码：第一行应返回行的python列表。我想看看第一个值：
mvv_list = mvv_count_df.select('mvv').collect()
firstvalue = mvv_list[0].getInt(0)
但是我收到第二行的错误消息：
AttributeError： getInt
看，为什么你这样做的方式不起作用。首先，您尝试从行类型中获取整数，集合的输出如下所示：
>>> mvv_list = mvv_count_df.select('mvv').collect()
>>> mvv_list[0]
Out: Row(mvv=1)
如果你采取这样的东西：
>>> firstvalue = mvv_list[0].mvv
Out: 1
您将获得mvv值。如果你想要数组的所有信息，你可以采取这样的东西：
>>> mvv_array = [int(row.mvv) for row in mvv_list.collect()]
>>> mvv_array
Out: [1,2,3,4]
但是，如果您对另一列尝试相同的操作，则会得到：
>>> mvv_count = [int(row.count) for row in mvv_list.collect()]
Out: TypeError: int() argument must be a string or a number, not 'builtin_function_or_method'
发生这种情况是因为count是一个内置方法。并且该列与count 同名。执行此操作的解决方法是将 count 的列名更改为 _count ：
>>> mvv_list = mvv_list.selectExpr("mvv as mvv", "count as _count")
>>> mvv_count = [int(row._count) for row in mvv_list.collect()]
但不需要此解决方法，因为您可以使用字典语法访问该列：
>>> mvv_array = [int(row['mvv']) for row in mvv_list.collect()]
>>> mvv_count = [int(row['count']) for row in mvv_list.collect()]
它最终会起作用！
后面一行给出你想要的列表。
mvv = mvv_count_df.select("mvv").rdd.flatMap(lambda x: x).collect()
这将以列表形式为您提供所有元素。
mvv_list = list(
mvv_count_df.select('mvv').toPandas()['mvv']
)
我进行了基准分析，list(mvv_count_df.select('mvv').toPandas()['mvv'])是最快的方法。 我很惊讶。
我使用 Spark 2.4.5 的 5 节点 i3.xlarge 集群(每个节点有 30.5 GB 的 RAM 和 4 个内核(在 10 万/1 亿行数据集上运行了不同的方法。 数据均匀分布在 20 个具有单列的快速压缩 Parquet 文件上。
以下是基准测试结果(以秒为单位的运行时间(：
+-------------------------------------------------------------+---------+-------------+
| Code | 100,000 | 100,000,000 |
+-------------------------------------------------------------+---------+-------------+
| df.select("col_name").rdd.flatMap(lambda x: x).collect() | 0.4 | 55.3 |
| list(df.select('col_name').toPandas()['col_name']) | 0.4 | 17.5 |
| df.select('col_name').rdd.map(lambda row : row[0]).collect()| 0.9 | 69 |
| [row[0] for row in df.select('col_name').collect()] | 1.0 | OOM |
| [r[0] for r in mid_df.select('col_name').toLocalIterator()] | 1.2 | * |
+-------------------------------------------------------------+---------+-------------+
* cancelled after 800 seconds
在驱动程序节点上收集数据时要遵循的黄金法则：
- 尝试用其他方法解决问题。 将数据收集到驱动程序节点的成本很高，不会利用 Spark 群集的强大功能，应尽可能避免。
- 收集尽可能少的行。 在收集数据之前聚合、删除重复数据、筛选和修剪列。 尽可能少地向驱动程序节点发送数据。
toPandas在Spark 2.3中得到了显著改进。 如果您使用的是早于 2.3 的 Spark 版本，这可能不是最佳方法。
有关更多详细信息/基准测试结果，请参阅此处。
在我的数据上，我得到了这些基准：
>>> data.select(col).rdd.flatMap(lambda x: x).collect()
0.52 秒
>>> [row[col] for row in data.collect()]
0.271 秒
>>> list(data.select(col).toPandas()[col])
0.427 秒
结果是一样的
以下代码将帮助您
mvv_count_df.select('mvv').rdd.map(lambda row : row[0]).collect()
一个可能的解决方案是使用 pyspark.sql.functions 中的 collect_list() 函数。这会将所有列值聚合到一个 pyspark 数组中，该数组在收集时会转换为 python 列表：
mvv_list = df.select(collect_list("mvv")).collect()[0][0]
count_list = df.select(collect_list("count")).collect()[0][0]
如果您收到以下错误：
属性错误："列表"对象没有属性"收集">
此代码将解决您的问题：
mvv_list = mvv_count_df.select('mvv').collect()
mvv_array = [int(i.mvv) for i in mvv_list]
让我们创建有问题的数据帧
df_test = spark.createDataFrame(
[
(1, 5),
(2, 9),
(3, 3),
(4, 1),
],
['mvv', 'count']
)
df_test.show()
这给了
+---+-----+
|mvv|count|
+---+-----+
| 1| 5|
| 2| 9|
| 3| 3|
| 4| 1|
+---+-----+
然后应用 rdd.flatMap(f(.collect(( 来获取列表
test_list = df_test.select("mvv").rdd.flatMap(list).collect()
print(type(test_list))
print(test_list)
这给了
<type 'list'>
[1, 2, 3, 4]
你可以先收集df，将返回行类型的列表
row_list = df.select('mvv').collect()
迭代行以转换为列表
sno_id_array = [ int(row.mvv) for row in row_list]
sno_id_array
[1,2,3,4]
使用平面图
sno_id_array = df.select("mvv").rdd.flatMap(lambda x: x).collect()
尽管有很多答案，但当您需要将列表与when和isin命令结合使用时，其中一些答案不起作用。生成平面值列表的最简单但有效的方法是使用列表推导和[0]来避免行名：
flatten_list_from_spark_df=[i[0] for i in df.select("your column").collect()]
另一种方法是使用熊猫数据框，然后使用list功能，但它不方便且有效。