我使用的是spark-sql-2.4.1v。在我的用例中，使用windowspec/feature使用 rank() 功能。我必须找到某些分区键的最新记录并按顺序排列 insertion_date .
它非常慢。这个Windows能装吗 rank() 是否可用于生产等级代码？或者有没有别的方法可以推荐？特别是为了提高性能。
请给我建议。
我目前正在使用以下代码：

Dataset<Row> data =  sqlContext.read.format("org.apache.spark.sql.cassandra")
     .option("spark.cassandra.connection.host", hosts)
     .options(Map( "table" -> "source_table", "keyspace" -> "calc")).load()
     .where(col("error").equalTo(lit(200)))
     .filter(col("insert_date").gt(lit("2015-01-01")))
     .filter(col("insert_date").lt(lit("2016-01-01")))
     .where(col("id").equalTo(lit(mId)))

说明计划

== Physical Plan ==

* (1) Project [cast(cast(unix_timestamp(insert_date#199, yyyy-MM-dd, Some(America/New_York)) as timestamp) as date) AS insert_date#500, 3301 AS id#399, create_date#201, company_id#202, ... 76 more fields]

+- *(1) Filter (((((((cast(error#263 as int) = 200) && (cast(insert_date#199 as string) >= 2018-01-01)) && (cast(insert_date#199 as string) <= 2018-01-01)) && isnotnull(id#200)) && isnotnull(insert_date#199)) && isnotnull(error#263)) && (id#200 = 3301))
   +- *(1) Scan org.apache.spark.sql.cassandra.CassandraSourceRelation@261d7ee2 [... 76 more fields] PushedFilters: [IsNotNull(id), IsNotNull(insert_date), IsNotNull(error), EqualTo(id,3301)], ReadSchema: struct<...

阅读

+-----------+------+
|partitionId| count|
+-----------+------+
|       1829|    29|
|       1959| 16684|
|        496|  3795|
|       2659|   524|
|       1591|    87|
|       2811|  2436|
|       2235|   620|
|       2563|   252|
|       1721|    12|
|        737|  1695|
|        858|   182|
|       2580| 73106|
|       3179|   694|
|       1460|    13|
|       1990|    66|
|       1522|   951|
|        540|    11|
|       1127|823084|
|       2999|     9|
|        623|  6629|
+-----------+------+
only showing top 20 rows

20/05/19 06:32:53 WARN ReaderCassandra:  Processed  started : 1589864993863  ended : 1589869973496 timetaken : 4979 s

val ws = Window.partitionBy("id").orderBy(desc("insert_date"),desc("update_date"));

Dataset<Row> ranked_data =  data.withColumn("rank",rank().over(ws))
    .where($"rank".===(lit(1)))
    .select("*")