Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Computación distribuida usando Python

Computación distribuida usando Python

La computación distribuída es un nuevo modelo de computación que surgió con el objetivo de resolver problemas de computación masiva donde diferentes máquinas trabajan en paralelo formando un clúster de computación.
En los últimos años han surgido diferentes frameworks como Apache Hadoop, Apache Spark y Apache Flink que permiten resolver este tipo de problemas donde tenemos datos masivos desde diferentes fuentes de datos.
Dentro del ecosistema de Python podemos destacar las librerías de Pyspark y Dask de código abierto que permiten la ejecución de tareas de forma paralela y distribuida en Python.
Entre los puntos a tratar podemos destacar:
Introducción a la computación distribuida
Comparando tecnologías de computación distribuida
Frameworks y módulos en Python para computación distribuida
Casos de uso en proyectos Big Data

jmortegac

June 18, 2023
Tweet

More Decks by jmortegac

Other Decks in Technology

Transcript

  1. Agenda • Introducción a la computación distribuida • Comparando tecnologías

    de computación distribuida • Frameworks y módulos en Python para computación distribuida • Casos de uso en proyectos Big Data
  2. Introducción a la computación distribuida • Grandes volúmenes de datos

    • Necesidad de procesar esos datos de manera rápida • Tipos de datos complejos
  3. Comparando tecnologías de computación distribuida Apache Hadoop Apache Spark Apache

    Flink Velocidad de procesamiento Más lento que Spark y Flink 100 veces más rápido que Hadoop Más rápido que Spark Modelo de programación Map Reduce RDD(Conjuntos de datos resilientes) Flujos de datos Procesamiento de datos Por lotes Por micro-lotes y secuencial* Por lotes Soporte Streaming N/A Spark Streaming Flink Streaming Soporte ML N/A Spark ML Flink ML
  4. Frameworks y módulos en Python para computación distribuida • Velocidad:

    procesamiento en memoria • Ejecución perezosa(Lazy execution) • Map-Reduce
  5. Frameworks y módulos en Python para computación distribuida from pyspark

    import SparkContext sc = SparkContext.getOrCreate() from pyspark.sql import SparkSession spark = SparkSession.builder.getOrCreate() from pyspark.sql import SparkSession spark = SparkSession.builder.getOrCreate() sc = spark.sparkContext
  6. Frameworks y módulos en Python para computación distribuida • RDD

    (Resilient Distributed Dataset) • Transformaciones: tras aplicar una transformación, obtenemos un nuevo y modificado RDD basado en el original. • Acciones: una acción consiste en aplicar una operación sobre un RDD y obtener un valor como resultado, que dependerá del tipo de operación.
  7. Frameworks y módulos en Python para computación distribuida lista =

    ['uno','dos','dos','tres','cuatro'] listardd = sc.parallelize(lista) listardd = sc.parallelize(lista,4) print(listardd.collect()) # Visualizar la colección RDD rdd_fichero = sc.textFile("/home/fichero.txt") rdd_text = sc.parallelize(['red', 'red', 'blue', 'green', 'green','yellow']) rdd_aux = rdd_text.map(lambda x: (x,1)) rdd_result = rdd_aux.reduceByKey(lambda x,y: x+y) print (rdd_result.collect()) [('blue', 1), ('green', 2), ('yellow', 1), ('red', 2)]
  8. Frameworks y módulos en Python para computación distribuida Sobre los

    RDD de pares, podemos realizar las siguientes transformaciones: • keys: devuelve las claves • values: devuelve los valores • mapValues: Aplica la función sobre los valores • flatMapValues: Aplica la función sobre los valores y los aplana.
  9. Frameworks y módulos en Python para computación distribuida listaTuplas =

    [('a',1), ('z',3), ('b',4), ('c',3), ('a',4)] rddTuplas= sc.parallelize(listaTuplas) claves = rddTuplas.keys() # ['a', 'z', 'b', 'c', 'a'] valores = rddTuplas.values() # [1, 3, 4, 3, 4] rddMapValues = rddTuplas.mapValues(lambda x: (x,x*2)) # [('a', (1, 2)), ('z', (3, 6)), ('b', (4, 8)), ('c', (3, 6)), ('a', (4, 8))] rddFMV = rddTuplas.flatMapValues(lambda x: (x,x*2)) # [('a', 1), # ('a', 2), # ('z', 3), # ('z', 6), # ('b', 4),
  10. Frameworks y módulos en Python para computación distribuida from pyspark.sql

    import SparkSession # Iniciamos Spark de manera local spark = (SparkSession .builder .master("local[*]") .appName("Hola Mundo") .getOrCreate()) # Paralelizamos una lista del 0 al 9 y computamos la suma de los números result = (spark .sparkContext .parallelize(range(10)) .reduce(lambda x, y: x + y)) # Imprimimos el resultado print(f"La suma es {result}") $ spark-submit script.py
  11. Frameworks y módulos en Python para computación distribuida • Lazy

    execution ◦ compute() ◦ persistent() ◦ head()
  12. Frameworks y módulos en Python para computación distribuida https://distributed.dask.org/en/stable/examples/word-count.html import

    hdfs3 from collections import defaultdict, Counter from distributed import Client from distributed.diagnostics.progressbar import progress hdfs = hdfs3.HDFileSystem('NAMENODE_HOSTNAME', port=NAMENODE_PORT) client = Client('SCHEDULER_IP:SCHEDULER:PORT') filenames = hdfs.glob('/tmp/enron/*/*') print(filenames[:5]) print(hdfs.head(filenames[0]))
  13. Frameworks y módulos en Python para computación distribuida https://distributed.dask.org/en/stable/examples/word-count.html def

    count_words(fn): word_counts = defaultdict(int) with hdfs.open(fn) as f: for line in f: for word in line.split(): word_counts[word] += 1 return word_counts counts = count_words(filenames[0]) print(sorted(counts.items(), key=lambda k_v: k_v[1], reverse=True)[:10]) future = client.submit(count_words, filenames[0]) counts = future.result() print(sorted(counts.items(), key=lambda k_v: k_v[1], reverse=True)[:10])
  14. Frameworks y módulos en Python para computación distribuida https://distributed.dask.org/en/stable/examples/word-count.html futures

    = client.map(count_words, filenames) len(futures) futures[:5] progress(futures) def top_items(d): items = sorted(d.items(), key=lambda kv: kv[1], reverse=True)[:10000] return dict(items) futures2 = client.map(top_items, futures) results = client.gather(iter(futures2)) all_counts = Counter() for result in results: all_counts.update(result) print(len(all_counts)) print(sorted(all_counts.items(), key=lambda k_v: k_v[1], reverse=True)[:10])
  15. Frameworks y módulos en Python para computación distribuida • Documentación

    https://docs.dask.org • Dask.distributed https://distributed.dask.org • Blog de desarrollo https://blog.dask.org • Algoritmos de Machine Learning implementados en Dask https://github.com/dask/dask-ml
  16. Frameworks y módulos en Python para computación distribuida Apache Spark

    Dask Modelo de programación Map Reduce Tasks DataFrames Spark DataFrame tiene su propia API y modelo de memoria Dask DataFrame reutiliza la API y el modelo de memoria de Pandas. Soporte ML Mllib Dask ML Soporte lenguajes Scala,Python,R Python,C,C++
  17. Casos de uso en proyectos Big Data from pyspark.sql import

    SparkSession spark = SparkSession \ .builder \ .appName("Streaming WordCount") \ .master("local[2]") \ .getOrCreate() # Creamos un flujo de escucha sobre netcat en localhost:9999 # En Spark Streaming, la lectura se realiza mediante readStream lineasDF = spark.readStream \ .format("socket") \ .option("host", "localhost") \ .option("port", "9999") \ .load() $ nc -lk 9999
  18. Casos de uso en proyectos Big Data # Leemos las

    líneas y las pasamos a palabras. # Sobre ellas, realizamos la agrupación count (transformación) from pyspark.sql.functions import explode, split palabrasDF = lineasDF.select(explode(split(lineasDF.value, ' ')).alias('palabra')) cantidadDF = palabrasDF.groupBy("palabra").count() # Mostramos las palabras por consola # En Spark Streaming, la persistencia se realiza mediante writeStream y en vez de realizar un save, ahora utilizamos start wordCountQuery = cantidadDF.writeStream \ .format("console") \ .outputMode("complete") \ .start() # dejamos Spark a la escucha para que cierre el el flujo cuando termine de procesar wordCountQuery.awaitTermination()
  19. Conclusiones • La distribución de la carga de trabajo entre

    los distintos nodos de un cluster en una solución Big Data, permite realizar las tareas en paralelo, consiguiendo grandes mejoras en la eficiencia de los procesos. • La computación distribuida es más eficiente en escenarios con unos cálculos a realizar sobre un gran conjunto de datos. De esta manera se paraleliza el trabajo y se reduce el tiempo total de cálculo.
  20. Conclusiones • A la hora de diseñar la arquitectura de

    un sistema Big Data, es conveniente tener en cuenta el grado de paralelización de las tareas, que será función del número de procesadores y cores disponibles en el sistema. Si el grado de paralelización es bajo, el rendimiento en según qué tareas, puede acabar siendo también muy bajo. • Con soluciones como PySpark y Dask podemos realizar cálculos y transformaciones sobre grandes volúmenes de datos, de forma distribuida sobre un cluster y trabajar con procesamiento en paralelo.