Есть ли способ пропустить/выбросить/игнорировать записи в Spark во время карты?

У нас есть очень стандартное задание Spark, которое считывает файлы журнала из s3 и затем обрабатывает их. Очень простой искр...

val logs = sc.textFile(somePathTos3)
val mappedRows = logs.map(log => OurRowObject.parseLog(log.split("\t")))
val validRows = mappedRows.filter(log => log._1._1 != "ERROR")
...and continue processing

Где OurRowObject.parseLine берет необработанную строку журнала и сопоставляет ее с некоторой (ключом, значением) парой (например, ( (1,2,3,4), (5,6,7) )), после чего мы можем обрабатывать. Теперь, если parseLine обнаруживает журнал проблемных проблем, empty и т.д.), он вернет некоторое значение дозорности (например, ( ("ERROR", ...), (...) ), который затем отфильтрует шаг фильтра.

Теперь, что я пытался найти способ сделать, это просто не включать строку проблемы во время карты... какой-то способ сказать искру "Эй, это пустая/неправильная строка, пропустите ее и не включают в себя пару для него", вместо этого дополнительного шага фильтра.

Мне еще не удалось найти способ сделать это, и мне очень интересно, что эта функциональность не существует (AFAICanFind).

Спасибо

Ответ 1

Вы можете заставить парсер вернуть опцию [Значение] вместо значения. Таким образом, вы можете использовать flatMap для сопоставления строк с строками и удаления недопустимых.

В грубых строках что-то вроде этого:

def parseLog(line:String):Option[Array[String]] = {
    val splitted = log.split("\t")
    if (validate(splitted)) Some(splitted) else None
}

val validRows = logs.flatMap(OurRowObject.parseLog(_))

Ответ 2

Один подход заключается в использовании однопараметрической перегрузки collect (вместо map или flatMap) и PartialFunction. Это немного сложно, если частичная функция, которая вам нужна, не является полностью тривиальной. На самом деле вы, вероятно, не будете потому, что вам нужно проанализировать и проверить, что я буду моделировать с двумя частичными функциями (хотя первый из них определяется для всех входов).

// this doesn't really need to be a partial function but we'll 
// want to compose it with one and end up with a partial function
val split: PartialFunction[String, Array[String]] = {
  case log => log.split("\t")
}

// this really needs to be a partial function
val validate: PartialFunction[Array[String], Array[String]] = {
  case lines if lines.length > 2 => lines
}

val splitAndValidate = split andThen validate

val logs = sc.parallelize(Seq("a\tb", "u\tv\tw", "a", "x\ty\tz"), 4)

// only accept the logs with more than two entries
val validRows = logs.collect(splitAndValidate)

Это отлично работает Scala, но это не работает, потому что splitAndValidate не является сериализуемым, и мы используем Spark. (Обратите внимание, что split и validate являются сериализуемыми: проблема кроется в составе!) Итак, нам нужно сделать PartialFunction сериализуемым:

class LogValidator extends PartialFunction[String, Array[String]] with Serializable {

  private val validate: PartialFunction[Array[String], Array[String]] = {
    case lines if lines.length > 2 => lines
  }

  override def apply(log: String) : Array[String] = {
    validate(log.split("\t"))
  }

  override def isDefinedAt(log: String) : Boolean = {
    validate.isDefinedAt(log.split("\t"))
  }

}

Тогда мы можем назвать

val validRows = logs.collect(new LogValidator())