Apache Spark 使用传递值映射和减少

Question

我在从 Cassandra 加载的 RDD 上有一个简单的 map 和 reduce 作业。 代码看起来像这样

sc.cassandraTable("app","channels").select("id").toArray.foreach((o) => {

  val orders = sc.cassandraTable("fam", "table")
    .select("date", "f2", "f3", "f4")
    .where("id = ?", o("id")) # This o("id") is the ID i want later append to the finished list

  val month = orders
    .map( oo => {
      var total_revenue = List(oo.getIntOption("f2"), oo.getIntOption("f3"), oo.getIntOption("f4")).flatten.reduce(_ + _)
      (getDateAs("hour", oo.getDate("date")), total_revenue)
    })
    .reduceByKey(_ + _)
})

所以这段代码总结了收入并返回类似的东西

(2014-11-23 18:00:00, 12412)
(2014-11-23 19:00:00, 12511)

现在我想将其保存回 Cassandra 表 revenue_hour，但我需要该列表中的 ID，类似这样。

(2014-11-23 18:00:00, 12412, "CH1")
(2014-11-23 19:00:00, 12511, "CH1")

我怎样才能使这项工作不仅仅是一个（键，值）列表？我怎样才能传递更多不应该转换的值，而只是传递到最后，以便我可以将其保存回 Cassandra？

Answer 1

也许您可以使用一个类并通过流程使用它。我的意思是，定义 RevenueHour 类

case class RevenueHour(date: java.util.Date,revenue: Long, id: String)

然后在 map 阶段构建一个中间 RevenueHour，然后在 reduce 阶段构建另一个。

val map: RDD[(Date, RevenueHour)] = orders.map(row => 
  (
    getDateAs("hour", oo.getDate("date")), 
    RevenueHour(
      row.getDate("date"),
      List(row.getIntOption("f2"),row.getIntOption("f3"),row.getIntOption("f4")).flatten.reduce(_ + _),
      row.getString("id")
    )
  )
).reduceByKey((o1: RevenueHour, o2: RevenueHour) => RevenueHour(getDateAs("hour", o1.date), o1.revenue + o2.revenue, o1.id))

我使用 o1 RevenueHour 因为 o1 和 o2 将具有相同的键和相同的 id（因为之前的 where 子句）。

希望对你有帮助。

Answer 2

针对该问题提出的方法是通过迭代一组 id 并仅在数据的一个（可能很小的）子集上应用 Spark 作业来对数据的处理进行排序。

在不知道“通道”和“表”数据之间的关系如何的情况下，我看到了两个充分利用 Spark 并行处理数据能力的选项：

选项 1

如果“表”表中的数据（从这里称为“订单”）包含我们在报告中需要的所有 ID 集，我们可以将报告逻辑应用于整个表：

根据问题，我们将使用这个 C* 架构：

CREATE TABLE example.orders (id text,
      date TIMESTAMP,
      f2 decimal,
      f3 decimal,
      f4 decimal,
      PRIMARY KEY(id, date)
);

通过提供一个表示表架构的案例类，访问 cassandra 数据变得更加容易：

case class Order(id: String, date:Long, f2:Option[BigDecimal], f3:Option[BigDecimal], f4:Option[BigDecimal]) {
    lazy val total = List(f2,f3,f4).flatten.sum
}

然后我们可以根据cassandra表定义一个rdd。当我们提供案例类作为类型时，spark-cassandra 驱动程序可以直接执行转换，以方便我们：

val ordersRDD = sc.cassandraTable[Order]("example", "orders").select("id", "date", "f2", "f3", "f4")

val revenueByIDPerHour = ordersRDD.map{order => ((order.id, getDateAs("hour", order.date)), order.total)}.reduceByKey(_ + _) 

最后保存回 Cassandra：

revenueByIDPerHour.map{ case ((id,date), revenue) => (id, date, revenue)}
    .saveToCassandra("example","revenue", SomeColumns("id", "date", "total"))

选项 2

如果应该使用 ("app","channels") 表中包含的 id 来过滤 id 集（例如有效 id），那么我们可以将此表中的 id 与订单连接。该工作将与上一个类似，但增加了：

val idRDD = sc.cassandraTable("app","channels").select("id").map(_.getString)
val ordersRDD = sc.cassandraTable[Order]("example", "orders").select("id", "date", "f2", "f3", "f4")
val validOrders = idRDD.join(ordersRDD.map(order => (id,order))

这两种方式说明了如何使用 Cassandra 和 Spark，利用 Spark 操作的分布式特性。它也应该比对“通道”表中的每个 ID 执行查询要快得多。