您可以将一个 Spark Dataframe 嵌套在另一个 Dataframe 中吗？
Can you nest a Spark Dataframe in another Dataframe?
在 spark 中，我希望能够并行处理多个数据帧。
我正在尝试的方法是将数据帧嵌套在父数据帧中，但我不确定语法或是否可能。
例如我有以下2个数据帧：
df1:
+-----------+---------+--------------------+------+
|id |asset_id | date| text|
+-----------+---------+--------------------+------+
|20160629025| A1|2016-06-30 11:41:...|aaa...|
|20160423007| A1|2016-04-23 19:40:...|bbb...|
|20160312012| A2|2016-03-12 19:41:...|ccc...|
|20160617006| A2|2016-06-17 10:36:...|ddd...|
|20160624001| A2|2016-06-24 04:39:...|eee...|
df2:
+--------+--------------------+--------------+
|asset_id| best_date_time| Other_fields|
+--------+--------------------+--------------+
| A1|2016-09-28 11:33:...| abc|
| A1|2016-06-24 00:00:...| edf|
| A1|2016-08-12 00:00:...| hij|
| A2|2016-07-01 00:00:...| klm|
| A2|2016-07-10 00:00:...| nop|
所以我想把这些结合起来产生这样的东西。
+--------+--------------------+-------------------+
|asset_id| df1| df2|
+--------+--------------------+-------------------+
| A1| [df1 - rows for A1]|[df2 - rows for A1]|
| A2| [df1 - rows for A2]|[df2 - rows for A2]|
请注意，我不想加入或联合它们，因为那样会非常稀疏（我实际上有大约 30 个数据框和数千个资产，每个资产都有数千行）。
然后我计划对此做一个 groupByKey 以便我得到类似这样的东西，我可以调用一个函数：
[('A1', <pyspark.resultiterable.ResultIterable object at 0x2534310>), ('A2', <pyspark.resultiterable.ResultIterable object at 0x25d2310>)]
我是 spark 的新手，非常感谢任何帮助。
TL;DR 无法嵌套 DataFrames 但可以使用复杂类型。
在这种情况下，例如（Spark 2.0 或更高版本）：
from pyspark.sql.functions import collect_list, struct
df1_grouped = (df1
.groupBy("asset_id")
.agg(collect_list(struct("id", "date", "text"))))
df2_grouped = (df2
.groupBy("asset_id")
.agg(collect_list(struct("best_date_time", "Other_fields"))))
df1_grouped.join(df2_grouped, ["asset_id"], "fullouter")
但你必须知道：
- 挺贵的。
- 它的应用有限。一般来说，嵌套结构使用起来很麻烦，并且需要复杂且昂贵的（尤其是在 PySpark 中）UDF。
在 spark 中，我希望能够并行处理多个数据帧。
我正在尝试的方法是将数据帧嵌套在父数据帧中，但我不确定语法或是否可能。
例如我有以下2个数据帧： df1:
+-----------+---------+--------------------+------+
|id |asset_id | date| text|
+-----------+---------+--------------------+------+
|20160629025| A1|2016-06-30 11:41:...|aaa...|
|20160423007| A1|2016-04-23 19:40:...|bbb...|
|20160312012| A2|2016-03-12 19:41:...|ccc...|
|20160617006| A2|2016-06-17 10:36:...|ddd...|
|20160624001| A2|2016-06-24 04:39:...|eee...|
df2:
+--------+--------------------+--------------+
|asset_id| best_date_time| Other_fields|
+--------+--------------------+--------------+
| A1|2016-09-28 11:33:...| abc|
| A1|2016-06-24 00:00:...| edf|
| A1|2016-08-12 00:00:...| hij|
| A2|2016-07-01 00:00:...| klm|
| A2|2016-07-10 00:00:...| nop|
所以我想把这些结合起来产生这样的东西。
+--------+--------------------+-------------------+
|asset_id| df1| df2|
+--------+--------------------+-------------------+
| A1| [df1 - rows for A1]|[df2 - rows for A1]|
| A2| [df1 - rows for A2]|[df2 - rows for A2]|
请注意，我不想加入或联合它们，因为那样会非常稀疏（我实际上有大约 30 个数据框和数千个资产，每个资产都有数千行）。
然后我计划对此做一个 groupByKey 以便我得到类似这样的东西，我可以调用一个函数：
[('A1', <pyspark.resultiterable.ResultIterable object at 0x2534310>), ('A2', <pyspark.resultiterable.ResultIterable object at 0x25d2310>)]
我是 spark 的新手，非常感谢任何帮助。
TL;DR 无法嵌套 DataFrames 但可以使用复杂类型。
在这种情况下，例如（Spark 2.0 或更高版本）：
from pyspark.sql.functions import collect_list, struct
df1_grouped = (df1
.groupBy("asset_id")
.agg(collect_list(struct("id", "date", "text"))))
df2_grouped = (df2
.groupBy("asset_id")
.agg(collect_list(struct("best_date_time", "Other_fields"))))
df1_grouped.join(df2_grouped, ["asset_id"], "fullouter")
但你必须知道：
- 挺贵的。
- 它的应用有限。一般来说，嵌套结构使用起来很麻烦，并且需要复杂且昂贵的（尤其是在 PySpark 中）UDF。