¿Cómo divido un RDD en dos o más RDD?

No es posible producir varios RDD a partir de una sola transformación*. Si desea dividir un RDD, debe aplicar un filter para cada condición de división. Por ejemplo:

def even(x): return x % 2 == 0
def odd(x): return not even(x)
rdd = sc.parallelize(range(20))

rdd_odd, rdd_even = (rdd.filter(f) for f in (odd, even))

Si solo tiene una condición binaria y la computación es costosa, puede preferir algo como esto: {[15]]}

kv_rdd = rdd.map(lambda x: (x, odd(x)))
kv_rdd.cache()

rdd_odd = kv_rdd.filter(lambda kv: kv[1]).keys()
rdd_even = kv_rdd.filter(lambda kv: not kv[1]).keys()

Significa solo un solo cálculo de predicados, pero requiere una transferencia adicional sobre todos los datos.

Es importante tener en cuenta que siempre y cuando un RDD de entrada se almacene correctamente en caché y no no hay suposiciones adicionales con respecto a la distribución de datos no hay diferencia significativa cuando se trata de la complejidad del tiempo entre el filtro repetido y el bucle for con if-else anidado.

Con N elementos y M condiciones el número de operaciones que tiene que realizar es claramente proporcional a N veces M. En el caso del bucle for, debe estar más cerca de (N + MN) / 2 y el filtro repetido es exactamente NM, pero al final del día no es nada más que O(NM). Puedes ver mi discusión* * con Jason Lenderman para leer sobre algunos pros y contras.

En el nivel muy alto usted debe considerar dos cosas: {[15]]}

1. Las transformaciones de chispa son perezosas, hasta que ejecuta una acción, su RDD no se materializa

   ¿por Qué importa? Volviendo a mi ejemplo:

   rdd_odd, rdd_even = (rdd.filter(f) for f in (odd, even))
   

   Si después decido que solo necesito rdd_odd entonces no hay razón para materializarme rdd_even.

   Si echa un vistazo a su ejemplo de SAS para calcular work.split2 necesita materializar ambos datos de entrada y work.split1.

2. Los RDD proporcionan una API declarativa. Cuando se utiliza filter o map depende completamente de Spark engine cómo se realiza esta operación. Siempre y cuando las funciones pasadas a las transformaciones estén libres de efectos secundarios, crea múltiples posibilidades para optimizar toda una tubería.

A fin de cuentas, este caso no es lo suficientemente especial como para justificar su propia transformación.

Este mapa con patrón de filtro se utiliza realmente en un núcleo Chispa. Ver mi respuesta a Cómo funciona Sparks RDD.randomSplit realmente divide el RDD y una parte relevante del método randomSplit.

Si el único objetivo es lograr una división en la entrada, es posible usar la cláusula partitionBy para DataFrameWriter qué formato de salida de texto:

def makePairs(row: T): (String, String) = ???

data
  .map(makePairs).toDF("key", "value")
  .write.partitionBy($"key").format("text").save(...)

* Solo hay 3 tipos básicos de transformaciones en Spark:

• RDD [T] = > RDD [T]
• RDD [T] = > RDD [U]
• (RDD[T], RDD[U]) => RDD[W]

Donde T, U, W pueden ser tipos atómicos o productos / tuplas (K, V). Cualquier otra operación tiene que ser expresada usando alguna combinación de lo anterior. Puede consultar el documento RDD original para más detalles.

** http://chat.stackoverflow.com/rooms/91928/discussion-between-zero323-and-jason-lenderman

* * * Véase también Scala Spark: ¿Dividir la colección en varios RDD?