Spark java.lang.UnsupportedOperationException: пустая коллекция

Question

Когда я запускаю этот код, в некоторых случаях я получаю пустую ошибку коллекции.

    val result = df
                  .filter(col("channel_pk") === "abc")
                  .groupBy("member_PK")
                  .agg(sum(col("price") * col("quantityOrdered")) as "totalSum")
                  .select("totalSum")
                  .rdd.map(_ (0).asInstanceOf[Double]).reduce(_ + _)

Ошибка происходит в этой строке:

.rdd.map(_ (0).asInstanceOf[Double]).reduce(_ + _)

Когда коллекция пуста, я хочу, чтобы result был равен 0. Как я могу это сделать?

Answer 1

Когда коллекция пуста, я хочу, чтобы результат был равен 0. Как я могу это сделать?

Перед агрегацией просто проверьте, есть ли в кадре данных строки или нет

val result = if(df.take(1).isEmpty) 0 else df
  .filter(col("channel_pk") === "abc")
  .groupBy("member_PK")
  .agg(sum(col("price") * col("quantityOrdered")) as "totalSum")
  .select("totalSum")
  .rdd.map(_(0).asInstanceOf[Double]).reduce(_ + _)

или вы можете использовать count тоже

val result = if(df.count() == 0) 0 else df
  .filter(col("channel_pk") === "abc")
  .groupBy("member_PK")
  .agg(sum(col("price") * col("quantityOrdered")) as "totalSum")
  .select("totalSum")
  .rdd.map(_(0).asInstanceOf[Double]).reduce(_ + _)

Answer 2

Ошибка появляется только в этой строке, потому что это первый раз, когда вы делаете какое-либо действие. до этого искра ничего не исполняет (лень). ваш df просто пуст Вы можете проверить это, добавив перед этим: assert(!df.take(1).isEmpty)