Spark:分组数据的"数量"是转换还是行动？

Question

我知道count调用RDD或DataFrame是一个动作.但是在摆弄火花壳时,我发现了以下情况

scala> val empDF = Seq((1,"James Gordon", 30, "Homicide"),(2,"Harvey Bullock", 35, "Homicide"),(3,"Kristen Kringle", 28, "Records"),(4,"Edward Nygma", 30, "Forensics"),(5,"Leslie Thompkins", 31, "Forensics")).toDF("id", "name", "age", "department")
empDF: org.apache.spark.sql.DataFrame = [id: int, name: string, age: int, department: string]

scala> empDF.show
+---+----------------+---+----------+
| id|            name|age|department|
+---+----------------+---+----------+
|  1|    James Gordon| 30|  Homicide|
|  2|  Harvey Bullock| 35|  Homicide|
|  3| Kristen Kringle| 28|   Records|
|  4|    Edward Nygma| 30| Forensics|
|  5|Leslie Thompkins| 31| Forensics|
+---+----------------+---+----------+

scala> empDF.groupBy("department").count //count returned a DataFrame
res1: org.apache.spark.sql.DataFrame = [department: string, count: bigint]

scala> res1.show
+----------+-----+                                                              
|department|count|
+----------+-----+
|  Homicide|    2|
|   Records|    1|
| Forensics|    2|
+----------+-----+

当我调用countGroupedData(empDF.groupBy("department"))时,我得到另一个DataFrame作为结果(res1).这让我相信count在这种情况下是一种转变.当我打电话时没有触发任何计算这一事实进一步支持了它count,而是在我跑的时候开始计算res1.show.

我一直无法找到任何count可能构成转换的文档.有人可以对此有所了解吗？

Answer 1

情况1：

您用来rdd.count()计算行数。由于它启动 DAG 执行并将数据返回给驱动程序，因此它是 RDD 的一个操作。

例如：rdd.count // it returns a Long value

案例2：

如果您在 Dataframe 上调用 count，它会启动 DAG 执行并将数据返回给驱动程序，这是 Dataframe 的一个操作。

例如：df.count // it returns a Long value

案例3：

在您的情况下，您正在调用groupBy返回dataframe对象RelationalGroupedDataset，并且您正在调用count返回 a 的分组数据集Dataframe，因此它是一个转换，因为它不会将数据获取到驱动程序并启动 DAG 执行。

例如：

 df.groupBy("department") // returns RelationalGroupedDataset
          .count // returns a Dataframe so a transformation
          .count // returns a Long value since called on DF so an action

Answer 2

在.count()你已经在你的代码中使用结束RelationalGroupedDataset,它创建与分组数据集的元素计数的新列.这是一种转变.参考:https: //spark.apache.org/docs/1.6.0/api/scala/index.html#org.apache.spark.sql.GroupedDataset

在.count()您通常在使用RDD/DataFrame/Dataset从上面的完全不同,这.count()是一个动作.参考:https://spark.apache.org/docs/1.6.0/api/scala/index.html#org.apache.spark.rdd.RDD

编辑:

总是在对distributedDataSet进行操作时使用.count(),.agg()以避免将来出现混淆:

empDF.groupBy($"department").agg(count($"department") as "countDepartment").show