java - 使用 Apache Spark 和 Java 对 DataFrame 进行分组和聚合?

标签 java apache-spark apache-spark-sql

我在 Spark 中加载了具有以下模式的 DataFrame:

电子邮件, 名, 姓, 订单号

我如何通过电子邮件对其进行分组,计算每组中的记录并返回具有此架构的 DataFrame:

电子邮件, 名, 姓, 订单数

最佳答案

这是在 Scala 中实现的一种方法:

val df = sc.parallelize(Seq(("a","b","c",1),("a","b","c",2),("x","xb","xc",3),("y","yb","yc",4),("x","xb","xc",5))).toDF("email","first_name","last_name","order_id")

df.registerTempTable("df")

sqlContext.sql("select * from (select email, count(*) as order_count from df group by email ) d1 join df d2 on d1.email = d2.email")

Java 中,考虑到您已经创建了 DataFrame,它实际上是相同的代码:

DataFrame results = sqlContext.sql("select * from (select email, count(*) as order_count from df group by email ) d1 join df d2 on d1.email = d2.email");

尽管如此,甚至认为这是直接的解决方案,但我认为这是一种不好的做法,因为您的代码将难以维护和发展。更清洁的解决方案是:

DataFrame email_count = df.groupBy("email").count();
DataFrame results2 = email_count.join(df, email_count.col("email").equalTo(df.col("email"))).drop(df.col("email"));

关于java - 使用 Apache Spark 和 Java 对 DataFrame 进行分组和聚合?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/34996378/

相关文章:

java - 如何克隆 Java 中的异常

java - java 中的格式化输出,如 ls

python - 使用spark SQL读取Parquet格式的不存在列

apache-spark - 有没有 hadoop map-reduce 比 apache spark 做得更好的用例?

apache-spark - 使用检查点 Spark Stream 的中流更改配置

apache-spark - pyspark dataframe 数组基于多个值时如何处理

python - Spark - 嵌套RDD操作

java - 在 Android 代码中,什么暗示了程序期望变量具有的特征?

java - 库存程序只读取一行?

python - 如何获得今天 - PySpark(SQL) 中的 “6 months” 日期