我有3个数据集,我想加入并进行分组以获取包含汇总数据的CSV。
数据作为拼合文件存储在Hadoop中,我正在使用Zeppelin运行Apache Spark + Scala进行数据处理。
我的数据集如下所示:
user_actions.show(10)
user_clicks.show(10)
user_options.show(10)
+--------------------+--------------------+
| id| keyword|
+--------------------+--------------------+
|00000000000000000001| aaaa1|
|00000000000000000002| aaaa1|
|00000000000000000003| aaaa2|
|00000000000000000004| aaaa2|
|00000000000000000005| aaaa0|
|00000000000000000006| aaaa4|
|00000000000000000007| aaaa1|
|00000000000000000008| aaaa2|
|00000000000000000009| aaaa1|
|00000000000000000010| aaaa1|
+--------------------+--------------------+
+--------------------+-------------------+
| search_id| selected_user_id|
+--------------------+-------------------+
|00000000000000000001| 1234|
|00000000000000000002| 1234|
|00000000000000000003| 1234|
|00000000000000000004| 1234|
+--------------------+-------------------+
+--------------------+----------+----------+
| search_id| user_id| position|
+--------------------+----------+----------+
|00000000000000000001| 1230| 1|
|00000000000000000001| 1234| 3|
|00000000000000000001| 1232| 2|
|00000000000000000002| 1231| 1|
|00000000000000000002| 1232| 2|
|00000000000000000002| 1233| 3|
|00000000000000000002| 1234| 4|
|00000000000000000003| 1234| 1|
|00000000000000000004| 1230| 1|
|00000000000000000004| 1234| 2|
+--------------------+----------+----------+
我想要实现的目标是为每个用户id获得带有关键字的JSON,因为我需要将它们导入MySQL并将user_id作为PK。
user_id,keywords
1234,"{\"aaaa1\":3.5,\"aaaa2\":0.5}"
如果JSON不是开箱即用的,我可以使用元组或任何字符串:
user_id,keywords
1234,"(aaaa1,0.58333),(aaaa2,1.5)"
到目前为止,我所做的是:
val user_actions_data = user_actions
.join(user_options, user_options("search_id") === user_actions("id"))
val user_actions_full_data = user_actions_data
.join(
user_clicks,
user_clicks("search_id") === user_actions_data("search_id") && user_clicks("selected_user_id") === user_actions_data("user_id"),
"left_outer"
)
val user_actions_data_groupped = user_actions_full_data
.groupBy("user_id", "search")
.agg("search" -> "count", "selected_user_id" -> "count", "position" -> "avg")
def udfScoreForUser = ((position: Double, searches: Long) => ( position/searches ))
val search_log_keywords = user_actions_data_groupped.rdd.map({row => row(0) -> (row(1) -> udfScoreForUser(row.getDouble(4), row.getLong(2)))}).groupByKey()
val search_log_keywords_array = search_log_keywords.collect.map(r => (r._1.asInstanceOf[Long], r._2.mkString(", ")))
val search_log_keywords_df = sc.parallelize(search_log_keywords_array).toDF("user_id","keywords")
.coalesce(1)
.write.format("csv")
.option("header", "true")
.mode("overwrite")
.save("hdfs:///Search_log_testing_keywords/")
虽然这对于小型数据集可以正常工作,但是我的输出CSV文件是:
user_id,keywords
1234,"(aaaa1,0.58333), (aaaa2,0.5)"
在处理200 + GB的数据时,我遇到了问题。
我对Spark&Scala相当陌生,但我想我缺少一些东西,我不应该使用DF到rdd,收集以映射到数组,然后将其并行化回DF以将其导出为CSV。
总结一下,我想对所有关键字应用评分,并按用户ID分组,然后将其保存到CSV。到目前为止,我所做的只是使用一个小的数据集,但是当我将其应用于200GB +的数据时,apache spark会失败。
最佳答案
是的,任何依赖Spark中collect
的东西通常都是错误的-除非您要调试某些东西。当您调用collect
时,所有数据都以阵列的形式收集在驱动程序中,因此对于大多数大数据集,这甚至都不是一种选择-您的驱动程序将抛出OOM而死。
我不明白的是,为什么首先要收藏?为什么不简单地映射到分布式数据集?
search_log_keywords
.map(r => (r._1.asInstanceOf[Long], r._2.mkString(", ")))
.toDF("user_id","keywords")
.coalesce(1)
.write.format("csv")
.option("header", "true")
.mode("overwrite")
.save("hdfs:///Search_log_testing_keywords/")
这样,所有事情都是并行进行的。
关于在
dataframes
和rdds
之间切换,那么我现在不必为此担心太多。我知道社区主要提倡使用dataframes
,但是根据Spark版本和您的用例,rdds
可能是一个更好的选择。关于scala - 使用hadoop Parquet将大数据处理为CSV输出,我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/44020646/