我有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/")

这样,所有事情都是并行进行的。

关于在dataframesrdds之间切换,那么我现在不必为此担心太多。我知道社区主要提倡使用dataframes,但是根据Spark版本和您的用例,rdds可能是一个更好的选择。

关于scala - 使用hadoop Parquet将大数据处理为CSV输出,我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/44020646/

10-14 18:08
查看更多