DataFrame Spark 的优化查询

Question

我尝试从 Hive 表创建 DataFrame。但我对 Spark API 的工作很糟糕。

我需要帮助优化方法 getLastSession 中的查询，将两个任务合二为一：

val pathTable = new File("/src/test/spark-warehouse/test_db.db/test_table").getAbsolutePath
val path      = new Path(s"$pathTable${if(onlyPartition) s"/name_process=$processName" else ""}").toString
val df        = spark.read.parquet(path)


def getLastSession: Dataset[Row] = {
  val lastTime        = df.select(max(col("time_write"))).collect()(0)(0).toString
  val lastSession     = df.select(col("id_session")).where(col("time_write") === lastTime).collect()(0)(0).toString
  val dfByLastSession = df.filter(col("id_session") === lastSession)

  dfByLastSession.show()
  /*
  +----------+----------------+------------------+-------+
  |id_session|      time_write|               key|  value|
  +----------+----------------+------------------+-------+
  |alskdfksjd|1639950466414000|schema2.table2.csv|Failure|

  */
  dfByLastSession
}

PS。我的源表（例如）：

name_process	id_session	time_write	key	value
OtherClass	jsdfsadfsf	43434883477	schema0.table0.csv	Success
OtherClass	jksdfkjhka	23212123323	schema1.table1.csv	Success
OtherClass	alskdfksjd	23343212234	schema2.table2.csv	Failure
ExternalClass	sdfjkhsdfd	34455453434	schema3.table3.csv	Success

Answer 1

您可以像这样在窗口中使用row_number：

import org.apache.spark.sql.expressions.Window

val dfByLastSession = df.withColumn(
  "rn", 
  row_number().over(Window.orderBy(desc("time_write")))
).filter("rn=1").drop("rn")
    
dfByLastSession.show()

但是，由于您没有按任何字段进行分区，因此可能会降低性能。

您可以在代码中更改的另一件事是使用结构排序来通过一个查询获取与最近的time_write 关联的id_session：

val lastSession = df.select(max(struct(col("time_write"), col("id_session")))("id_session")).first.getString(0)

val dfByLastSession = df.filter(col("id_session") === lastSession)