spark中的dataframe查找每天的最大行程

olhwl3o2 于 2021-05-27 发布在 Hadoop

关注(0)|答案(1)|浏览(439)

环境：Java1.8，vm cloudera快速启动。我从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
   .........
   .........

所以，我每天都想找出每个供应商（a和b）拥有最多公交线路的时间。使用java和spark。结果可能是：

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..)
....

mu java代码是：

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();

但是我找不到哪一个小时是每天最多的路线。

Java hadoop apache-spark

来源：https://stackoverflow.com/questions/59604881/find-max-trips-per-day-from-dataframe-in-java-spark

1条答案

按热度按时间

hc8w905p1#

我对java不太熟悉，所以我试着用scala来解释它。
找出每个供应商每天的最大路线小时数的关键是按 (vendor, day, hour) ，然后按 (vendor, day) 计算各组最大cnt对应的小时数。这个 day 以及 hour 每个记录的 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."))

赞(0）回复(0）举报 2021-05-27

我来回答

spark中的dataframe查找每天的最大行程

1条答案

相关问题

热门标签

最新问答