我使用的是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("*")
暂无答案!
目前还没有任何答案,快来回答吧!