代码之家 › 专栏 › 技术社区 › ScalaBoy

如何获取列的最小值和最大值?

apache-spark-sql apache-spark scala

ScalaBoy · 技术社区 · 6 年前

我想对我的代码进行概念检查。目标是计算字段的最小值 minTimestamp 以及字段的最大值 maxTimestamp 在数据帧中 df ,并删除所有其他值。例如:

测向

src    dst  minTimestamp   maxTimestamp
1      3    1530809948     1530969948
1      3    1540711155     1530809945
1      3    1520005712     1530809940
2      3    1520005712     1530809940

答案应如下:

结果:

src    dst  minTimestamp   maxTimestamp
1      3    1520005712     1530969948
2      3    1520005712     1530809940

这是我的代码:

val cw_min = Window.partitionBy($"src", $"dst").orderBy($"minTimestamp".asc)
val cw_max = Window.partitionBy($"src", $"dst").orderBy($"maxTimestamp".desc)

val result = df
  .withColumn("rn", row_number.over(cw_min)).where($"rn" === 1).drop("rn")
  .withColumn("rn", row_number.over(cw_max)).where($"rn" === 1).drop("rn")

可以用吗 Window 像我在代码示例中那样按顺序运行? 问题是我总是得到 最小时间戳 和 最大时间戳 .

2 回复 | 直到 6 年前

-2

botchniaque 6 年前

val spark: SparkSession = ???
df.createOrReplaceTempView("myDf")
val df2 = spark.sql("""
    select 
      src, 
      dst, 
      min(minTimestamp) as minTimestamp, 
      max(maxTimestamp) as maxTimestamp 
    from myDf group by src, dst""")

val df2 = df
   .groupBy("src", "dst")
   .agg(min("minTimestamp"), max("maxTimestamp"))

Leo C 6 年前

groupBy min max

import org.apache.spark.sql.functions._

val df = Seq(
  (1, 3, 1530809948L, 1530969948L),
  (1, 3, 1540711155L, 1530809945L),
  (1, 3, 1520005712L, 1530809940L),
  (2, 3, 1520005712L, 1530809940L)
).toDF("src", "dst", "minTimestamp", "maxTimestamp")

df.groupBy("src", "dst").agg(
    min($"minTimestamp").as("minTimestamp"), max($"maxTimestamp").as("maxTimestamp")
  ).
  show
// +---+---+------------+------------+                                             
// |src|dst|minTimestamp|maxTimestamp|
// +---+---+------------+------------+
// |  2|  3|  1520005712|  1530809940|
// |  1|  3|  1520005712|  1530969948|
// +---+---+------------+------------+

推荐文章

Geoffrey · Pyspark:将数据帧保存到多个具有单个文件特定大小的镶木地板文件中

1 年前

Bruno Peixoto · Spark群集CI管道构建失败

1 年前

codebot · 将df从pandas转换为PySpark时会删除列名

1 年前

mcsilvio · 在foreach中组织联接的最佳方式是什么?

1 年前

Dhruv · 在sbt控制台上运行Spark

1 年前

Leonard · Pyspark:JSON到Pyspark数据帧

1 年前

billie class · 将列中的值重写为列表中的下一个值

2 年前

Calcutta · Google Colab中的Spark SQL在大数据上失败

2 年前

Doraemon · PySpark:使用不同值的字符串类型列创建聚合列

2 年前

OdiumPura · 使用JDBC(Sql server)查询tempview

2 年前