Как передать два значения как State в Spark Streaming? - PullRequest
Новая
       37

Как передать два значения как State в Spark Streaming?

0 голосов
/ 06 октября 2018

Я попытался реализовать приложение Spark Streaming, которое считывает потоковые данные из Kafka.Потоковые данные представляют собой пары (ключ, значение) в форме «String, int», и я хочу вычислить среднее значение каждого ключа.Данные в форме, как показано ниже: 

x,20
y,10
z,3
...

Я хочу измерить среднее значение для каждого ключа в состоянии с сохранением состояния.Поэтому я намереваюсь сохранить сумму значений и сколько раз соответствующий ключ появляется в State в функции отображения.

 def mappingFunc(key: String, value: Option[Double], state: State[Double], count: State[Int]): (String, Double) = {
        val sum = value.getOrElse(0.0) + state.getOption.getOrElse(0.0)
        val cnt = count.getOption.getOrElse(1) + 1
        state.update(sum)
        count.update(cnt)
        val output = (key, sum/cnt)
        output
    }

Это напоминает мне об ошибке: 

[error] /Users/Rabbit/Desktop/KTH_Second_Year/Periods/P1/Data-intensive_Computing/Lab_Assignment/lab3/src/sparkstreaming/KafkaSpark.scala:78: wrong number of type parameters for overloaded method value function with alternatives:
[error]   [KeyType, ValueType, StateType, MappedType](mappingFunction: org.apache.spark.api.java.function.Function3[KeyType,org.apache.spark.api.java.Optional[ValueType],org.apache.spark.streaming.State[StateType],MappedType])org.apache.spark.streaming.StateSpec[KeyType,ValueType,StateType,MappedType] <and>
[error]   [KeyType, ValueType, StateType, MappedType](mappingFunction: org.apache.spark.api.java.function.Function4[org.apache.spark.streaming.Time,KeyType,org.apache.spark.api.java.Optional[ValueType],org.apache.spark.streaming.State[StateType],org.apache.spark.api.java.Optional[MappedType]])org.apache.spark.streaming.StateSpec[KeyType,ValueType,StateType,MappedType] <and>
[error]   [KeyType, ValueType, StateType, MappedType](mappingFunction: (KeyType, Option[ValueType], org.apache.spark.streaming.State[StateType]) => MappedType)org.apache.spark.streaming.StateSpec[KeyType,ValueType,StateType,MappedType] <and>
[error]   [KeyType, ValueType, StateType, MappedType](mappingFunction: (org.apache.spark.streaming.Time, KeyType, Option[ValueType], org.apache.spark.streaming.State[StateType]) => Option[MappedType])org.apache.spark.streaming.StateSpec[KeyType,ValueType,StateType,MappedType]

Как я могу одновременно передать сумму значения и счет в Spark Streaming?

1 Ответ

0 голосов
/ 06 октября 2018

Вам нужно объединить сумму и количество как кортеж (Double, Int), который хранится в состоянии.Следующий фрагмент должен сделать трюк: 

def mappingFunc(key: String, value: Option[Double], state: State[(Double, Int)]): (String, Double) = {
    val (sum, cnt) = state.getOption.getOrElse((0.0, 0))
    val newSum = value.getOrElse(0.0) + sum
    val newCnt = cnt + 1
    state.update((newSum, newCnt))
    (key, newSum/newCnt)
}
...