从 Spark 中的 Cassandra 的第一行和最后一行减去值

Question

我有这段代码，它从Cassandra获取RDD，然后为每个键提取第一行和最后一行并减去它们。

val rdd = sc.cassandraTable("keyspace","table")
    .select("column1", "column2", "column3", "column4","column5")
    .as((i:String, p:String, e:String, c:Double, a:java.util.Date) => ((i), (c, a, p, e)))
    .groupByKey.mapValues(v => v.toList)
    .cache
val finalValues = rdd.mapValues(v => v.head)
val initialValues = rdd.mapValues(v => v.last)
val valuesCombined = finalValues.join(initialValues)
val results = valuesCombined.map(v => (v._2._1._1 - v._2._2._1))

在性能方面是否良好，还是有更好的解决方案？我不确定是否在内存中缓存整个数据集。

Answer 1

groupByKey打乱数据，不再保证分组值的顺序。它也相当昂贵。

如果您真的想在不DataFrames RDDs进行操作，并且订购基于您可以使用aggregateByKey的日期：

import scala.math.Ordering
type Record = (String, String, String, Double, java.util.Date)
val RecordOrd = Ordering.by[Record, java.util.Date](_._5)
val minRecord = ("", "", "", 0.0, new java.util.Date(Long.MinValue))
val maxRecord = ("", "", "", 0.0, new java.util.Date(Long.MaxValue))
def minMax(x: (Record, Record), y: (Record, Record)) = {
  (RecordOrd.min(x._1, y._1), RecordOrd.max(x._2, y._2))
}
rdd.aggregateByKey((maxRecord, minRecord))(
  (acc, x) => minMax(acc, (x, x)),
  minMax
)

使用DataFrames您可以尝试这样的事情：

import org.apache.spark.sql.functions.{col, lag, lead, when, row_number, max}
import org.apache.spark.sql.expressions.Window
val partition = Seq("column1")
val order = Seq("column5")
val columns = Seq("column2", "column3", "column4","column5")
val w = Window
  .partitionBy(partition.head, partition.tail: _*)
  .orderBy(order.head, order.tail: _*)
// Lead / lag of row number to mark first / last row in the group
val rn_lag = lag(row_number.over(w), 1).over(w)
val rn_lead = lead(row_number.over(w), 1).over(w)
// Select value if first / last row in the group otherwise null
val firstColumns = columns.map(
  c => when(rn_lag.isNull, col(c)).alias(s"${c}_first"))
val lastColumns = columns.map(
  c => when(rn_lead.isNull, col(c)).alias(s"${c}_last"))
// Add columns with first / last vals
val expanded = df.select(
  partition.map(col(_)) ++ firstColumns ++ lastColumns: _*)
// Aggregate to drop nulls
val aggExprs = expanded.columns.diff(partition).map(c => max(c).alias(c))
expanded.groupBy(partition.map(col(_)): _*).agg(aggExprs.head, aggExprs.tail: _*)

还有其他一些方法可以使用DataFrames解决此问题，包括通过structs订购和DataSet API。查看我对 SPARK 数据帧的回答：选择每个组的第一行

Answer 2

首先 - 我假设all变量指的是名为 rdd 的变量？创建后，您不需要使用 join（这在性能方面是昂贵的），您可以简单地将每个元素直接映射到您需要的结果：

val results = all.mapValues(v => v.head - v.last).values

现在 - 由于我们只在 RDD 上执行了一个操作，我们也可以摆脱cache() .