Scala - как получить и распечатать содержимое Either?

Question

Я обрабатываю данные, где некоторые записи могут быть повреждены.Поэтому я решил исследовать данные и использовал Either для разделения действительных и недействительных записей.

Я выяснил, как подсчитать количество записей каждого вида и теперь получаю выходные данные для failedCount и * 1005.* успешно.

Но у меня проблема с распечаткой каждой недействительной (левой) записи о продаже.Что может быть не так с моим подходом?

Я не получаю вывод при печати failedSales

  def filterSales(rawSales: RDD[Sale]): RDD[(String, Sale)] = {

    val filteredSales = rawSales
      .map(sale => {
        val saleOption = Try(sale.id -> sale)
        saleOption match {
          case Success(successSale) => Right(successSale)
          case Failure(e) => Left(s"Corrupted sale: $sale;", e)
        }
      })
    val failedCount: Long = filteredSales.filter(_.isLeft).count()
    val successCount: Long = filteredSales.filter(_.isRight).count()
    println("FAILED SALES COUNT: " + failedCount)
    println("SUCCESS SALES COUNT: " + successCount)

    // Problem here
    val failedSales: RDD[Either.LeftProjection[(String, Throwable), (String, Sale)]] = filteredSales.map(_.left)
    println("FAILED SALES: ")
    // Doesn't produce any output 
    failedSales.foreach(println)
  }

Answer 1

Когда вы вызываете foreach(fn) в СДР, функция fn (println в вашем случае) выполняется на подчиненных узлах, где СДР распространяется.Таким образом, это происходит где-то , но не в программе-драйвере, которую вы просматриваете.

Если у вас небольшой набор данных, вы можете collect() СДР, чтобы данные возвращались на ваш сервер.драйвер, и вы можете println, что.

Если у вас есть большие данные, вы могли бы saveAsTextFile(), чтобы они записывались в HDFS, и вы можете скачать оттуда.