在Java-Spark中从DataFrame查找每天最大行程

Question

环境:Java1.8，VM Cloudera QuickStart。我有从csv文件到Hadoop hdfs的数据。每一行代表一条公共汽车路线。

    id      vendor    start_datetime   end_datetime    trip_duration_in_sec
    17534   A         1/1/2013 12:00   1/1/2013 12:14  840
    68346   A         1/1/2013 12:13   1/1/2013 12:18  300
    09967   B         1/1/2013 12:34   1/1/2013 12:39  300
    09967   B         1/1/2013 12:44   1/1/2013 12:51  420
    09967   A         1/1/2013 12:54   1/1/2013 12:56  120
   .........
   .........

1/1/2013 (Day 1) - Vendor A has 3 bus routes at 12:00-13:00 hour. (That time 12:00-13:00, vendor A had the most bus routes..)
1/1/2013 (Day 1) - Vendor B has 2 bus routes at 12:00-13:00 hour. (That time 12:00-13:00, vendor B had the most bus routes..)
....

import static org.apache.spark.sql.functions;
import static org.apache.spark.sql.Row;

Dataset<Row> ds;
ds.groupBy(functions.window(col("start_datetime"), "1 hour").count().show();

Answer 1

我对Java不太熟悉，所以我试着用Scala来解释它。

找出每个供应商每天最大路由的小时数的关键是按(vendor,day,hour)计数，然后按(vendor,day)聚合，计算每组最大cnt对应的小时数。每个记录的日和小时可以由start_datetime解析。

val df = spark.createDataset(Seq(
("17534","A","1/1/2013 12:00","1/1/2013 12:14",840),
("68346","A","1/1/2013 12:13","1/1/2013 12:18",300),
("09967","B","1/1/2013 12:34","1/1/2013 12:39",300),
("09967","B","1/1/2013 12:44","1/1/2013 12:51",420),
("09967","A","1/1/2013 12:54","1/1/2013 12:56",120)
)).toDF("id","vendor","start_datetime","end_datetime","trip_duration_in_sec")

df.rdd.map(t => {
    val vendor = t(1)
    val day = t(2).toString.split(" ")(0)
    val hour = t(2).toString.split(" ")(1).split(":")(0)
    ((vendor, day, hour), 1)
})
// count by key
.aggregateByKey(0)((x: Int, y: Int) =>x+y, (x: Int, y: Int) =>x+y) 
.map(t => {
    val ((vendor, day, hour), cnt) = t;
    ((vendor, day), (hour, cnt))
})
// solve the max cnt by key (vendor, day)
.foldByKey(("", 0))((z: (String, Int), i: (String, Int)) => if (i._2 > z._2) i else z)
.foreach(t => println(s"${t._1._2} - Vendor ${t._1._1} has ${t._2._2} bus routes from ${t._2._1}:00 hour."))