Предотвратите больше ввода-вывода с несколькими конвейерами на одном и том же RDD

например. если я запускаю те же RDD чисел, где один поток фильтрует четные числа и усредняет их, а другие фильтры для нечетных и суммирует их. Если я напишу это как два конвейера над тем же RDD, это создаст два исполнения, которые будут сканировать RDD дважды, что может быть дорогостоящим с точки зрения ввода-вывода.

Как можно уменьшить этот IO только для чтения данных без перезаписи логики в один конвейер? Рамка, которая берет два конвейера и объединяет их в один, в порядке, конечно, до тех пор, пока разработчики продолжают работать над каждым конвейером независимо (в реальном случае эти конвейеры загружаются из отдельных модулей)

Дело не в использовании кеша() для достижения этого

Ответ 1

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

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

в хранилище данных памяти (например, Ignite heenenee)
ускоренное хранилище, например Alluxio

также неприемлемы. Это означает, что вам нужно найти некоторые из них для управления самим трубопроводом.

Хотя множественные преобразования могут быть сжаты вместе, каждое преобразование создает новый RDD. Это, в сочетании с вашим выражением о кэшировании, устанавливает относительно сильные ограничения на возможные решения.

Начнем с самого простого случая, когда все конвейеры могут быть выражены одноступенчатыми заданиями. Это ограничивает наш выбор отображением только заданий и простых сокращений по карте (например, описанных в вашем вопросе). Такие трубопроводы можно легко выразить как последовательность операций на локальных итераторах. Итак, следующее

import org.apache.spark.util.StatCounter

def isEven(x: Long) = x % 2 == 0
def isOdd(x: Long) = !isEven(x)

def p1(rdd: RDD[Long]) = {
  rdd
    .filter(isEven _)
    .aggregate(StatCounter())(_ merge _, _ merge _)
    .mean
}

def p2(rdd: RDD[Long]) = {
  rdd
    .filter(isOdd _)
    .reduce(_ + _)
}

может быть выражена как:

def p1(rdd: RDD[Long]) = {
  rdd
    .mapPartitions(iter => 
      Iterator(iter.filter(isEven _).foldLeft(StatCounter())(_ merge _)))
    .collect
    .reduce(_ merge _)
    .mean
}


def p2(rdd: RDD[Long]) = {
  rdd
    .mapPartitions(iter => 
      Iterator(iter.filter(isOdd _).foldLeft(0L)(_ + _)))
    .collect
    .reduce(_ + _)
    // identity _
}

В этот момент мы можем переписать отдельные задания следующим образом:

def mapPartitions2[T, U, V](rdd: RDD[T])(f: Iterator[T] => U, g: Iterator[T] => V)  = {
  rdd.mapPartitions(iter => {
    val items = iter.toList
    Iterator((f(items.iterator), g(items.iterator)))
  })
}

def reduceLocally2[U, V](rdd: RDD[(U, V)])(f: (U, U) => U, g: (V, V) => V) = {
   rdd.collect.reduce((x, y) => (f(x._1, y._1), g(x._2, y._2)))
}

def evaluate[U, V, X, Z](pair: (U, V))(f: U => X, g: V => Z) = (f(pair._1), g(pair._2)) 

val rdd = sc.range(0L, 100L)

def f(iter: Iterator[Long]) = iter.filter(isEven _).foldLeft(StatCounter())(_ merge _)
def g(iter: Iterator[Long]) = iter.filter(isOdd _).foldLeft(0L)(_ + _)


evaluate(reduceLocally2(mapPartitions2(rdd)(f, g))(_ merge _, _ + _))(_.mean, identity)

Самая большая проблема здесь заключается в том, что мы должны с готовностью оценивать каждый раздел, чтобы иметь возможность применять отдельные конвейеры. Это означает, что общие требования к памяти могут быть значительно выше по сравнению с той же самой логикой, применяемой отдельно. Без кеширования * это также бесполезно в случае многоступенчатых заданий.

Альтернативным решением является обработка данных по элементам, но обработка каждого элемента в виде кортежей seqs:

def map2[T, U, V, X](rdd: RDD[(Seq[T], Seq[U])])(f: T => V, g: U => X) = {
  rdd.map{ case (ts, us) => (ts.map(f), us.map(g)) }
}


def filter2[T, U](rdd: RDD[(Seq[T], Seq[U])])(
    f: T => Boolean, g: U => Boolean) = {
  rdd.map{ case (ts, us) => (ts.filter(f), us.filter(g)) }
}


def aggregate2[T, U, V, X](rdd: RDD[(Seq[T], Seq[U])])(zt: V, zu: X)
    (s1: (V, T) => V, s2: (X, U) => X, m1: (V, V) => V, m2: (X, X) => X) = {
  rdd.mapPartitions(iter => {
    var accT = zt
    var accU = zu
    iter.foreach { case (ts, us) => {
      accT = ts.foldLeft(accT)(s1)
      accU = us.foldLeft(accU)(s2)
    }}

    Iterator((accT, accU))
  }).reduce { case ((v1, x1), (v2, x2)) => ((m1(v1, v2), m2(x1, x2))) }
}

С API, подобным этому, мы можем выразить исходные конвейеры как:

val rddSeq = rdd.map(x => (Seq(x), Seq(x)))


aggregate2(filter2(rddSeq)(isEven, isOdd))(StatCounter(), 0L)(
  _ merge _, _ + _, _ merge _, _ + _
)

Этот подход немного более мощный, чем первый (вы можете легко реализовать некоторое подмножество методов byKey, если это необходимо), а требования к памяти в типичных конвейерах должны быть сопоставимы с основным API, но это также значительно более навязчиво.

* Вы можете проверить ответ, предоставленный eje для примеров мультиплексирования.