Apache Spark Scala API：Scala 中的 ReduceByKeyAndWindow

Question

由于我是 Spark 的 Scala API 新手，我遇到了以下问题：

在我的 java 代码中，我做了一个 reduceByKeyAndWindow 转换，但现在我看到，只有一个 reduceByWindow（因为 Scala 中也没有 PairDStream）。但是，我现在开始使用 Scala 的第一步：

import org.apache.hadoop.conf.Configuration;
import [...]

val serverIp = "xxx.xxx.xxx.xxx"
val receiverInstances = 2
val batchIntervalSec = 2
val windowSize1hSek = 60 * 60
val slideDurationSek = batchIntervalSec

val streamingCtx = new StreamingContext(sc, Seconds(batchIntervalSec))

val hadoopConf = sc.hadoopConfiguration
hadoopConf.set("fs.s3n.impl", "org.apache.hadoop.fs.s3native.NativeS3FileSystem")
hadoopConf.set("fs.s3.impl", "org.apache.hadoop.fs.s3native.NativeS3FileSystem")
hadoopConf.set("fs.s3n.awsAccessKeyId", "xxx")
hadoopConf.set("fs.s3n.awsSecretAccessKey", "xxx")

// ReceiverInputDStream
val receiver1 = streamingCtx.socketTextStream(serverIp, 7777)
val receiver2 = streamingCtx.socketTextStream(serverIp, 7778)

// DStream
val inputDStream = receiver1.union(receiver2)

// h.hh.plug.ts.val
case class DebsEntry(house: Integer, household: Integer, plug: Integer, ts: Long, value: Float)

// h.hh.plug.val
case class DebsEntryWithoutTs(house: Integer, household: Integer, plug: Integer, value: Float)

// h.hh.plug.1
case class DebsEntryWithoutTsCount(house: Integer, household: Integer, plug: Integer, count: Long)

val debsPairDStream = inputDStream.map(s => s.split(",")).map(s => DebsEntry(s(6).toInt, s(5).toInt, s(4).toInt, s(1).toLong, s(2).toFloat)) //.foreachRDD(rdd => rdd.toDF().registerTempTable("test"))

val debsPairDStreamWithoutDuplicates = debsPairDStream.transform(s => s.distinct())

val debsPairDStreamConsumptionGreater0 = debsPairDStreamWithoutDuplicates.filter(s => s.value > 100.0)

debsPairDStreamConsumptionGreater0.foreachRDD(rdd => rdd.toDF().registerTempTable("test3"))

val debsPairDStreamConsumptionGreater0withoutTs = debsPairDStreamConsumptionGreater0.map(s => DebsEntryWithoutTs(s.house, s.household, s.plug, s.value))

// 5.) Average per Plug
// 5.1) Create a count-prepared PairDStream (house, household, plug, 1)
val countPreparedPerPlug1h = debsPairDStreamConsumptionGreater0withoutTs.map(s => DebsEntryWithoutTsCount(s.house, s.household, s.plug, 1))

// 5.2) ReduceByKeyAndWindow
val countPerPlug1h = countPreparedPerPlug1h.reduceByWindow(...???...)

在步骤 5.1 之前一切正常。在 5.2 中，我现在想总结 countPreparedPerPlug1h 的 1，但前提是其他属性（房屋、家庭、插头）相等。 - 目标是获得每个（房屋、家庭、插头）组合的条目计数。有人可以帮忙吗？谢谢！

编辑 - 第一次尝试

我在步骤 5.2 中尝试了以下操作：

// 5.2)
val countPerPlug1h = countPreparedPerPlug1h.reduceByKeyAndWindow((a,b) => a+b, Seconds(windowSize1hSek), Seconds(slideDurationSek))

但在这里我得到以下错误：

<console>:69: error: missing parameter type
   val countPerPlug1h = countPreparedPerPlug1h.reduceByKeyAndWindow((a,b) => a+b, Seconds(windowSize1hSek), Seconds(slideDurationSek))
                                                                     ^

好像我使用reduceByKeyAndWindow转换错误，但错误在哪里？求和的值的类型为 Int，参见上面步骤 5.1 中的 countPreparedPerPlug1h。

Answer 1

您可以在 Scala 中使用reduceByKeyAndWindow，这比在您的 Java 版本中更简单。您没有 PairDStream，因为对是隐式确定的，您可以直接调用对方法。隐式解析转到PairDStreamFunctions

例如：

val myPairDStream: DStream[KeyType, ValueType] = ...
myPairDStream.reduceByKeyAndWindow(...)

真正的幕后花絮如下：

new PairDStreamFunctions(myPairDStream).reduceByKeyAndWindow(...)

PairDStreamFunctions 的这个包装器被添加到由 Tuple2 组成的任何 DStream 中

Answer 2

我知道了，现在似乎可以使用以下代码：

val countPerPlug1h = countPreparedPerPlug1h.reduceByKeyAndWindow({(x, y) => x + y}, {(x, y) => x - y}, Seconds(windowSize1hSek), Seconds(slideDurationSek))

感谢您的线索，@贾斯汀·皮奥尼