Question 1

¿Qué es Apache Spark?

Accepted Answer

Apache Spark es un sistema de computación distribuido de código abierto basado en Hadoop. 
Utilizada principalmente para el análisis y procesamiento de datos en los campos de Big Data, Machine Learning y procesamiento en tiempo real.

Answer

True

Question 2

¿Cuáles son las características más notables de Apache Spark?

Accepted Answer

velocidad, compatibilidad multi-formato y librerías integradas.
-Velocidad: su diseño se ha enfocado en optimizar el rendimiento en el procesamiento de datos a gran escala, aprovechando conceptos como el procesamiento en memoria y otras optimizaciones. Es una 100 veces más rápido que Hadoop.
-Facilidad de uso: dispone de APIs sencillas de utilizar para trabajar con grandes conjuntos de datos.
-Motor unificado: viene empaquetado con bibliotecas de nivel superior, que incluyen soporte para consultas SQL, transmisión de datos, aprendizaje automático y procesamiento de gráficos.

Answer

true

Question 3

¿Qué es "SCC"?

Accepted Answer

SCC, acrónimo de “Spark Cassandra Connector”. Es una herramienta que Apache Spark utiliza para acceder a la información (Datos) ubicados en diferentes bases de datos Cassandra.

Answer

true

Question 4

¿Sabes qué es Apache Hadoop

Accepted Answer

Hadoop nace como iniciativa de Apache para dar soporte al paradigma de programación Map-Reduce, que fue inicialmente publicado por Google.

El propósito principal del framework es almacenar grandes cantidades de datos y permitir consultas sobre dichos datos, que se ofrecerán con un bajo tiempo de respuesta. Esto se consigue mediante la ejecución distribuida de código en múltiples nodos (máquinas), cada uno de los cuales se encarga de procesar una parte del trabajo a realizar.

Answer

true

Question 5

¿Qué es MapReduce?

Accepted Answer

MapReduce es un framework que proporciona un sistema de procesamiento de datos paralelo y distribuido. Su nombre se debe a las funciones principales que son Map y Reduce, las cuales explicaremos a continuación. MapReduce está pensado para la solución práctica de algunos problemas que pueden ser paralelizados, pero se ha de tener en cuenta que no todos los problemas pueden resolverse eficientemente con MapReduce.

Answer

true

Question 6

¿Qué es "RDD"?

Accepted Answer

Resilient Distributed Dataset , Conjunto de datos distribuidos y resistentes , Es la estructura de datos fundamental de Apache Spark y es una colección na colección de elementos inmutable , tolerante a fallos y que es capaz de operar en paralelo.

Answer

true

Question 7

¿Qué es "inmutabilidad"?

Accepted Answer

cuando un elemento es inmutable, no puede ser cambiado o alterado de ninguna forma una vez se ha creado y asignado un valor.

Answer

true

Question 8

¿Qué es YARN?

Accepted Answer

YARN, es una de las características principales de Apache Spark. Principalmente se encarga de la administración de recursos pero también es utilizado para operar a través de diversos conjuntos Spark, esto se debe a su escalabilidad.

Answer

true

Question 9

¿Cuál es el lenguaje de programación más utilizado en Spark?

Accepted Answer

Aunque existen muchos desarrolladores que les gusta utilizar Python, Scala sigue siendo el lenguaje de programación más utilizado para Apache Spark.

Answer

true

Question 10

¿Cuántos administradores de clústeres se encuentran disponibles en Spark?

Accepted Answer

3 , YARN, Apache Mesos y despliegues independientes.

Answer

true

Question 11

¿Cuáles son las responsabilidades del motor Spark?

Accepted Answer

Generalmente, el motor Apache Spark es el encargado de establecer, extender (Distribuir) y luego monitorear los varios conjuntos de datos a lo largo de diferentes clústeres.

Answer

true

Question 12

Pregunta 10: ¿Qué son las "evaluaciones perezosas"?

Accepted Answer

este tipo de evaluaciones se retrasa hasta el punto en que se necesita el valor del elemento para ser empleado. Además, las evaluaciones perezosas, solo son ejecutadas una vez, no hay repetición de evaluaciones.

Answer

true

Question 13

¿Qué son las "particiones"?

Accepted Answer

Una partición, es una pequeña parte proveniente de un trozo más grande de datos. Las particiones están basadas en la lógica, son utilizados en Spark para administrar los datos a fin de lograr utilizar el mínimo de recursos en una red.

Answer

true

Question 14

¿Qué es la integración continua?

Accepted Answer

na práctica de ingeniería de software que consiste en hacer integraciones automáticas de un proyecto lo más a menudo posible para así poder detectar fallos cuanto antes. Entendemos por integración la compilación y ejecución de pruebas de todo un proyecto.

Answer

true

Question 15

Qué es Jenkins?

Accepted Answer

Jenkins es un servidor automatizado de integración continua de código abierto capaz de organizar una cadena de acciones que ayudan a lograr el proceso de integración continua (y mucho más) de manera automatizada.

Jenkins está completamente escrito en Java y es una aplicación conocida y reconocida por DevOps de todo el mundo, más de 300.000 instalaciones y + 15.500 estrellas en Github lo respaldan.

La razón por la que Jenkins se hizo tan popular es porque se encarga de supervisar las tareas repetitivas que surgen dentro del desarrollo de un proyecto.

Por ejemplo, si tu equipo está desarrollando un proyecto, Jenkins probará continuamente las compilaciones del mismo y será capaz de mostrarte los errores que aparezcan a lo largo de las primeras etapas del desarrollo.

Answer

true

Question 16

Que es SonarQube

Accepted Answer

Es una plataforma para evaluar la calidad del código fuente, realizando un análisis estático sobre dicho código, con el objetivo de advertirnos sobre diferentes puntos a mejorar y obtener métricas que nos ayudan a mejorar nuestro código.
El análisis estático corresponde al proceso de evaluar un software sin ejecutarlo, debo mencionar que si pretendemos evaluar la cobertura de los test estos se deben ejecutar.
En los aspectos que SonarQube nos ayuda son los siguientes:
Código duplicado
Código muerto
Estándares de codificación
Bugs
Complejidad ciclomática
Comentarios
Tests unitarios y de integración
Cobertura de código (Coverage)

Answer

true

Question 17

¿Para qué se utiliza el Apache Spark Streaming?

Accepted Answer

Spark Streaming, es el responsable de los procesos de transmisión de datos escalables e ininterrumpidos. Es una extensión principal del programa Spark y comúnmente utilizada por desarrolladores y programadores de Big Data.

Answer

true

Question 18

¿Es normal ejecutar todos tus procesos en un nodo localizado?

Accepted Answer

No, no lo es. De hecho, este es uno de los más comunes errores que un desarrollador Spark comete, especialmente cuando recién comienzan. Siempre debes tratar de distribuir tu flujo de datos, esto te permitirá acelerar y mejorar el proceso.

Answer

true

Question 19

¿Para qué se utiliza "SparkCore"?

Accepted Answer

SparkCore, es el motor principal responsable de todos los procesos que ocurren dentro de Spark. Teniendo esto en cuenta, probablemente no te sorprendería saber que posee un montón de tareas, entre ellas, monitoreo, administración de memoria y almacenaje, programación de tareas, etc.

Answer

true

Question 20

¿La API del sistema de archivos tiene un uso en Apache Spark?

Accepted Answer

Claro que si, Esta API en particular le permite a Spark leer y crear los datos a través de diferentes áreas de almacenamiento (Dispositivos).

Answer

true

Question 21

1. Apache Spark Core

Accepted Answer

Todas las funcionalidades proporcionadas por Apache Spark están construidas en la parte superior de Spark Core. Ofrece velocidad al proporcionar capacidad de cálculo en memoria. Por lo tanto, Spark Core es la base del procesamiento paralelo y distribuido de grandes conjuntos de datos.

Answer

true

Question 22

2. Apache Spark SQL

Accepted Answer

El componente Spark SQL es un marco distribuido para el procesamiento de datos estructurados. Usando Spark SQL, Spark obtiene más información sobre la estructura de datos y el cálculo. Con esta información, Spark puede realizar una optimización adicional. Utiliza el mismo motor de ejecución mientras calcula una salida. No depende de la API / lenguaje para expresar el cálculo.

Answer

true

Question 23

3. Apache Spark Streaming

Accepted Answer

Es un complemento del núcleo de Spark API que permite el procesamiento de transmisiones de datos en vivo escalables, de alto rendimiento y tolerantes a fallas. Spark puede acceder a datos de fuentes como Kafka, Flume, Kinesis o socket TCP. Puede operar usando varios algoritmos. Finalmente, los datos así recibidos se entregan al sistema de archivos, bases de datos y paneles en vivo. Spark utiliza Micro-lote para la transmisión en tiempo real.

Visite este enlace para obtener un estudio más detallado de los componentes del ecosistema de Apache Spark con sus características: Apache Spark: componentes del ecosistema .

Visite este enlace para obtener un estudio más detallado de los componentes del ecosistema de Apache Spark con sus características: Apache Spark: componentes del ecosistema .

Answer

true

Question 24

¿Cómo funciona Apache Spark?

Accepted Answer

Funciona con el sistema para distribuir datos a través del clúster y procesar los datos en paralelo. Spark utiliza arquitectura maestro / esclavo, es decir, un coordinador central y muchos trabajadores distribuidos. Aquí, el coordinador central se llama conductor.

El controlador se ejecuta en su propio proceso Java. Estos controladores se comunican con un número potencialmente grande de trabajadores distribuidos llamados ejecutores. Cada ejecutor es un proceso java separado. Una aplicación Spark es una combinación de controlador y sus propios ejecutores. Con la ayuda del administrador de clústeres, se lanza una aplicación Spark en un conjunto de máquinas. Standalone Cluster Manager es el administrador de clúster integrado predeterminado de Spark. Además de su administrador de clúster incorporado, Spark también funciona con algunos administradores de clúster de código abierto como Hadoop Yarn, Apache Mesos, etc.

Answer

true

Question 25

Apache Spark tiene los siguientes componentes:

Accepted Answer

Spark Core: Motor base para procesamiento de datos en paralelo y distribuido a gran escala. Cualquier mejora en el Core, es una mejora en las otros componentes.
Spark Streaming: Se utiliza para procesar datos en transmisión en tiempo real.
Spark SQL: Integra el procesamiento relacional con la programación funcional de Spark y, es una de las mas usadas.
GraphX: Gráficos y computación en paralelo a gráficos.
MLlib: Realiza el “Machine Learning” en Apache Spark

Answer

true

Question 26

¿Hay varios patrones de implementación para Spark, cada modo funciones?

Accepted Answer

Local: comúnmente utilizado para las pruebas de desarrollo local, local también se divide en un multithreading local de un solo roscado y local;

Standalone: ​​Modelo maestro / esclavo típico, Spark es compatible con ZookeEPER para implementar maestro ha;

En el hilo (modo de clúster): Ejecutar en el marco del explorador de hilos, mediante hilo, administración de recursos, Spark es responsable de la programación y cálculo de la tarea;

En Mesos: Ejecutar en el Marco de Mesos Explorer, por Mesos responsable de la gestión de recursos, Spark es responsable de la programación y el cálculo de la tarea;

En la nube (modo de clúster): por ejemplo, el EC2 de AWS, el uso de este modo puede acceder fácilmente a S3 de Amazon, SPROP Soporta múltiples sistemas de almacenamiento distribuidos: HDFS y S3, etc.

Answer

true

Question 27

¿Qué niveles de almacenamiento tienen Spark?

Accepted Answer

1) Memory_only: los datos se guardan en la memoria, y si la memoria no es suficiente, los datos pueden no durar.

2) Memory_and_disk: Los datos están preventivos en la memoria, y si la memoria no es suficiente, se guardará.

3) Memory_only_Ser: Similar a Memory_Only, la diferencia es serializar los datos en RDD, lo que ahorra memoria;

4) Memory_and_disk_ser: Similar a Memory_and_disk, la diferencia es serializar los datos en el RDD, lo que ahorra más memoria.

5) disk_only: escriba los datos a los archivos de disco;

6) Memory_only_2, Memory_and_disk_2, y similares: este sufijo _2, representa la copia de cada datos persistentes, y guarda la copia a otros nodos. Este mecanismo de persistencia basado en copasas se utiliza principalmente para la tolerancia a fallas.

Answer

true

Question 28

¿Qué es DataFrame?

Accepted Answer

DataFrame es un conjunto de datos distribuidos basado en RDD, similar a las tablas bidimensionales de las bases de datos tradicionales.

Answer

¿Qué es DataFrame?

Question 29

¿Cuál es la principal diferencia entre DataFrame y RDD?

Accepted Answer

El DataFrame tiene metainformación del esquema, es decir, cada columna del conjunto de datos de la tabla bidimensional representada por el DataFrame tiene un nombre y un tipo. Esto permite que SparkSQL obtenga información sobre información más estructural, optimizando así la fuente de datos oculta detrás del DataFrame y las transformaciones que actúan sobre el DataFrame, y finalmente logrando el objetivo de mejorar en gran medida la eficiencia del tiempo de ejecución. A diferencia de RDD, debido a que no hay forma de conocer la estructura interna específica de los elementos de datos almacenados, Spark Core solo puede realizar una optimización simple y general de la canalización a nivel de etapa.

Answer

true

Question 30

Funciones de DataFrame

Accepted Answer

2. Admite varios formatos de datos y varios sistemas de almacenamiento
3. Optimización avanzada para generar código a través del optimizador Catalyst
4. Integre sin problemas la infraestructura y las herramientas de big data convencionales a través de Spark
5. La API es compatible con los lenguajes Python, Java, Scala y R

Answer

Funciones de DataFrame

Question 31

Conceptos básicos de RDD

Accepted Answer

Cliente: El proceso del cliente es responsable de enviar trabajos al Máster.
El nodo maestro en el modo Maestro: Independiente es responsable de recibir los trabajos enviados por el Cliente, administrar al Trabajador e indicarle al Trabajador que comience a asignar los recursos del Controlador e inicie los recursos del Ejecutor.
Trabajador: el demonio en el nodo esclavo en el modo autónomo, responsable de administrar los recursos del nodo, reportar latidos al maestro regularmente, recibir comandos del maestro e iniciar el controlador y el ejecutor.
Controlador: cuando se ejecuta un trabajo de Spark, incluye un proceso de controlador, que también es el proceso principal del trabajo. Es responsable del análisis del trabajo, la generación de etapas y la programación de tareas para el Ejecutor. Incluyendo DAGScheduler, TaskScheduler.
Ejecutor: El lugar donde se ejecuta realmente el trabajo. Un clúster generalmente contiene múltiples Ejecutores. Cada Ejecutor recibe el comando del controlador Iniciar tarea. Un Ejecutor puede ejecutar una o más tareas.

Answer

true

Question 32

Términos comunes de RDD

Accepted Answer

DAGScheduler: Realice la descomposición de trabajos Spark en una o más etapas, cada etapa determina la cantidad de tareas de acuerdo con la cantidad de particiones RDD, y luego genera un conjunto de tareas correspondiente y lo coloca en TaskScheduler.
TaskScheduler: Realice la asignación de tareas al Ejecutor para su ejecución.
Tarea: la unidad de trabajo que se ejecuta en Executor
Trabajo: acción específica enviada por SparkContext, que a menudo corresponde a la acción
Etapa: cada trabajo se dividirá en muchos grupos de tareas (tarea), cada grupo de tareas se llama Etapa, también llamado TaskSet
RDD: abreviatura de Resilient Distributed Datasets, conjunto de datos distribuidos elásticos, es el módulo principal y la clase de Spark
Transformación / Acción: dos tipos de SparkAPI; el valor de retorno de la transformación sigue siendo un RDD, el valor de retorno de la acción no es menor que un RDD, sino una colección de Scala; todas las transformaciones son perezosas Estrategia, si solo envía la Transformación, el cálculo no se ejecutará. El cálculo solo se activará cuando se envíe la Acción.
DataFrame: RDD con información de esquema, que es principalmente datos estructurados muy abstractos.
DataSet: combinando las ventajas de DataFrame y RDD, no solo permite a los usuarios manipular convenientemente los objetos de dominio, sino que también tiene el rendimiento eficiente del motor de ejecución SQL.

Answer

true

Question 33

RDD proporciona dos tipos de operaciones

Accepted Answer

transformación y acción
1. La transformación consiste en obtener un nuevo RDD de muchas formas, como generar un nuevo RDD a partir de la fuente de datos y generar un nuevo RDD a partir del RDD.
2. La acción es obtener un valor o un resultado (almacenar en caché el RDD directamente en la memoria)
3. Todas las transformaciones son estrategias perezosas, es decir, si la transformación solo se envía, el cálculo no se ejecutará y el cálculo solo se activará cuando se envíe la acción

Answer

true

Question 34

La diferencia entre transformación y acción en RDD

Accepted Answer

La transformación generará un nuevo RDD, y este último solo devolverá el resultado de una operación en el RDD al programa, pero no generará un nuevo RDD; no importa cuántas operaciones de transformación se realicen, el RDD realmente no realizará la operación (registro de linaje), La operación solo se activará cuando se ejecute la operación de acción.

Answer

true

Question 35

¿Cuál es la mayor diferencia entre RDD y DSM?

Accepted Answer

DSM(distributed shared memory)
RDD solo se puede crear mediante una conversión de grano grueso, mientras que DSM permite leer y escribir datos en cada ubicación de memoria. Bajo esta definición, DSM incluye no solo los sistemas tradicionales de memoria compartida, sino también Piccolo y bases de datos distribuidas que proporcionan DHT (tabla hash distribuida) compartida.

Answer

true

Question 36

¿Cuáles son las ventajas de RDD?

Accepted Answer

1. Mecanismo eficiente de tolerancia a fallas
2. Mitigación del problema de los nodos rezagados (mitigar rezagados):
3. Operación por lotes:
4. Degradar con elegancia

Answer

true

Question 37

9. ¿Cómo obtener RDD?

Accepted Answer

1. Obtenga del sistema de archivos compartidos (como HDFS)
2. Convierta el RDD existente
3. Ponga en paralelo la colección scala existente (siempre que sea un objeto Seq) llamando al método de paralelización de SparkContext
4. Cambie la persistencia de los RDD existentes; los RDD son perezosos y de corta duración.

Answer

true

Question 38

Dos tipos de RDD que dependerán de

Accepted Answer

Dependencias estrechas (dependencias estrechas) y dependencias amplias (dependencias amplias).
Dependencia estrecha significa que cada partición del RDD principal solo es utilizada por una partición del RDD secundario. En consecuencia, la amplia dependencia significa que la partición del RDD principal depende de las particiones de varios RDD secundarios. Por ejemplo, el mapa es una dependencia estrecha y la combinación conducirá a una dependencia amplia
Características de la clasificación de dependencia:
Primero, la dependencia estrecha puede calcular directamente un cierto dato correspondiente al RDD secundario en un determinado nodo informático mediante el cálculo de un determinado dato del RDD principal;
En segundo lugar, cuando se pierden datos, solo es necesario volver a calcular la parte que falta para recuperar la dependencia estrecha;

Answer

true

Question 39

Principios básicos de Spark Streaming

Accepted Answer

El principio básico de Spark Streaming es dividir el flujo de datos de entrada en unidades de segmentos de tiempo (segundos) y luego procesar los datos de cada segmento de tiempo de una manera por lotes

Answer

true

Question 40

Case 1Potencia una lista y salida

Accepted Answer

val input = sc.parallelize(List(1,2,3,4))
 val result = input.map(x => x*x)
 println(result.collect().mkString(","))

Answer

true

Question 41

Case 3  Imprimir elementos rdd

Accepted Answer

rdd.foreach (println) o rdd.map (println).
rdd.collect().foreach(println）
rdd.take(100).foreach(println）

Answer

true

Question 42

Case 2 wordcount

Accepted Answer

val textFile = sc.textFile(args(1)) 
 val result = textFile.flatMap(line => line.split("\s+")).map(word => (word, 1)).reduceByKey(_ + _)
 println(result.collect().mkString(","))
 result.saveAsTextFile(args(2))

Answer

true

Question 43

Que es having

Accepted Answer

La función HAVING se utiliza para incluir condiciones con alguna función SQL del tipo SUM, MAX, ..
Como la cláusula WHERE no se puede utilizar con funciones SQL del tipo SUM, MAX, entonces utilizamos en su lugar, HAVING.

Sintaxis HAVING

SELECT columna1, SUM(columna2)
FROM tabla
GROUP BY columna1
HAVING SUM(columna2) < número

Answer

true

Question 44

Diferencia entre Groupby, distinc y having

Accepted Answer

Utilice DISTINCT si solo desea eliminar duplicados. Utilice GROUPY BY si desea aplicar operadores agregados (MAX, SUM, GROUP_CONCAT, ..., o una cláusula HAVING).

Answer

true

Question 45

Que es 1.Flume

Accepted Answer

Flume es un sistema de recolección de registros distribuido
, con funciones como alta confiabilidad, alta disponibilidad, administración de transacciones y reinicio fallido. La velocidad de procesamiento de datos es rápida y se puede utilizar en entornos de producción.
El núcleo de Flume es el agente.
El agente es un proceso de Java que se ejecuta en el lado de la recopilación de registros, recibe registros a través del agente, los almacena temporalmente y los envía al destino.
El agente contiene 3 componentes principales: fuente, canal y receptor.
El componente de origen está dedicado a recopilar registros y puede procesar varios tipos de datos de registro en varios formatos, incluidos avro, thrift, exec, jms, spoolingdirectory, netcat, sequencegenerator, syslog, http, legado, personalizado. Una vez que el componente de origen recopila los datos, se almacenan temporalmente en el canal.
El componente Channel está dedicado a almacenar temporalmente datos en el agente, que pueden almacenarse en memoria, jdbc, archivo o personalizado. Los datos del canal solo se eliminarán después de que el receptor se envíe correctamente.
El componente Sink se utiliza para enviar datos al destino. El destino incluye hdfs, logger, avro, thrift, ipc, file, null, hbase, solr y custom.

Answer

true

Question 46

¿Qué es un motor de búsqueda?

Accepted Answer

Un motor de búsqueda se refiere a un sistema que recopila información de Internet de acuerdo con ciertas estrategias y utiliza programas informáticos específicos, organiza y procesa la información, proporciona servicios de recuperación a los usuarios y muestra información relevante relacionada con la recuperación de usuarios. Los motores de búsqueda incluyen índice de texto completo, índice de catálogo, meta motor de búsqueda, motor de búsqueda vertical, motor de búsqueda colectiva, motor de búsqueda de portal y lista de enlaces gratuitos.

Answer

true

Question 47

¿Qué es Lucene?

Accepted Answer

Lucene es una biblioteca de búsqueda de información escalable y de alto rendimiento, es decir, no es un motor de búsqueda de texto completo completo, sino una arquitectura de motor de búsqueda completa, que proporciona un motor de consulta completo, un motor de indexación y un motor de análisis de texto parcial.

Answer

true

Question 48

¿Qué es Elasticsearch?

Accepted Answer

Elasticsearch es una herramienta de análisis y búsqueda de texto completo de código abierto altamente escalable.
Le permite almacenar, buscar y analizar rápidamente datos de gran capacidad casi en tiempo real. Elasticsearch es un motor de búsqueda de código abierto basado en ApacheLucene (TM). Ya sea en campos de código abierto o propietarios, Lucene puede considerarse como la biblioteca de motores de búsqueda más avanzada, de mejor rendimiento y más funcional hasta el momento.
ElasticSearch tiene 4 formas de crear una base de datos
La forma más fácil es usar indexAPI para enviar un documento a un índice específico, que generalmente se implementa a través de curltools.
El segundo y tercer método son a través de bulkAPI y UDPbulkAPI. La diferencia entre los dos está solo en el método de conexión.
La cuarta forma es a través de un río enchufable. River se ejecuta en ElasticSearch y puede importar datos de bases de datos externas a ES. Cabe señalar que la construcción de datos solo se realiza en fragmentos, no en réplicas.

Answer

true

Question 49

¿Qué es Hive?

Accepted Answer

Hive es una herramienta de almacenamiento de datos basada en Hadoop, que puede asignar archivos de datos estructurados a una tabla de base de datos y proporcionar funciones de consulta similares a SQL. La esencia es convertir HQL al programa MapReduce

Answer

true

Question 50

Importar librerias en scala

Accepted Answer

import org.apache.spark.SparkConf 
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD

Answer

true

Question 51

Crea un nuevo RDD con 3 particiones a partir del siguiente array de números. Calcula:

val array = Array(23,54,32,66,0,-14,-8,10)

Valor máximo y mínimo. Suma de los valores.

Accepted Answer

val array = Array(23,54,32,66,0,-14,-8,10)
val numerosRDD = sc.parallelize(array,3)
array: Array[Int] = Array(23, 54, 32, 66, 0, -14, -8, 10)
numerosRDD: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:29
println("El valor máximo dentro de numerosRDD es " + numerosRDD.max() + " y el mínimo " + numerosRDD.min())
El valor máximo dentro de numerosRDD es 66 y el mínimo -14
println("La suma de todos los elementos es: " + numerosRDD.reduce((x,y) => x+y))
La suma de todos los elementos es: 163

Answer

true

Question 52

Haciendo uso del fichero “LorenIpsum.txt”, carga su contenido en un RDD. Luego, implementa un cuenta palabras. Muestra por pantalla las 10 palabras con más apariciones.

Accepted Answer

//carga del fichero loremipsum.txt
val lorem = sc.textFile("./documentos/loremipsum.txt")

//dividir por palabras
val palabras = lorem.flatMap(line => line.split(" "))

//map para contar crear el par clave-valor para cada palabra
.map(word => (word,1))

//numero de ocurrencias por cada clave
.reduceByKey(_+_)

//ordenación por Value y salida con los 10 resultados con más coincidencias.
//(_._2, false) Ordena por el segundo valor y descendentemente
palabras.sortBy(_._2, false).take(10)
lorem: org.apache.spark.rdd.RDD[String] = ./documentos/loremipsum.txt MapPartitionsRDD[2] at textFile at <console>:29
palabras: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[5] at reduceByKey at <console>:38
res2: Array[(String, Int)] = Array((in,6), ("",4), (dolor,4), (dolore,4), (ut,3), (voluptate,2), (deserunt,2), (ea,2), (eu,2), (laboris,2))

Answer

true

Question 53

Realiza un filtrado del RDD anterior quedándonos únicamente con las palabras de tamaño 6 o superior. Luego, transforma esas palabras a mayúsculas. Por último, para una muestra de 25% palabras, imprime por pantalla: “La palabra xxx tiene tamaño YY”

Accepted Answer

val mayor6 = lorem.flatMap(line => line.split(" ")).filter(p => p.length() >= 6)
val maymayor6 = mayor6.map(x => x.toUpperCase())

//función para la selección de elementos, sample(remplazo : bool, cantidad : double, semilla : int) )
val sample =mayor6.sample(false, 0.25, 245)
sample.foreach(p => println("La palabra " + p + " tiene tamaño " + p.length()))
La palabra eiusmod tiene tamaño 7
La palabra dolore tiene tamaño 6
La palabra ullamco tiene tamaño 7
La palabra voluptate tiene tamaño 9
La palabra dolore tiene tamaño 6
La palabra Excepteur tiene tamaño 9
La palabra cupidatat tiene tamaño 9
La palabra proident, tiene tamaño 9
La palabra consectetur tiene tamaño 11
La palabra adipiscing tiene tamaño 10
La palabra aliqua. tiene tamaño 7
La palabra veniam, tiene tamaño 7
La palabra aliquip tiene tamaño 7
La palabra fugiat tiene tamaño 6
La palabra Excepteur tiene tamaño 9
La palabra proident, tiene tamaño 9
La palabra officia tiene tamaño 7

mayor6: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[12] at filter at <console>:30
maymayor6: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[13] at map at <console>:31
sample: org.apache.spark.rdd.RDD[String] = PartitionwiseSampledRDD[14] at sample at <console>:34

Answer

true

Question 54

Dataframe y Datasets

Accepted Answer

Answer

true

Question 55

Filtrar los tweets en español.

Accepted Answer

tweetSpanish: Unit = ()

Answer

true

Question 56

Obtener los hashtags tranding topic

Accepted Answer

Obtener los hashtags tranding topic

//ahora se va a seleccionar los hastags de los tweets.
//hay muchos tweets sin hastag por lo que se procede a copiar los hastags en un array y posteriormente se seleccionan los elementos del array que no estén vacios.
val hashtag = tweetsJson.select("entities.hashtags.text").as[Array[String]]
hashtag.show()
+--------------------+
| text|
+--------------------+
| []|
| []|
| []|
| []|
|[nra, tcot, WAR, ...|
| []|
| []|
| []|
| []|
| []|
| []|
| []|
| []|
| []|
| []|
| []|
| []|
| []|
| []|
|[reasonsobamaislate]|
+--------------------+
only showing top 20 rows

hashtag: org.apache.spark.sql.Dataset[Array[String]] = [text: array<string>]
//se eliminan las posiciones vacias
val h = hashtag.filter(p => p.size > 0)
h.show()
+--------------------+
| text|
+--------------------+
|[nra, tcot, WAR, ...|
|[reasonsobamaislate]|
| [bias, Nra]|
| [Israel, Gaza]|
|[gun, PresidentOb...|
| [Pakistan]|
| [GAY]|
| [NYC]|
| [Syria]|
|[FiscalCliff, MC,...|
| [imagine]|
| [MUITO, TRISTE]|
| [Pakistan, C]|
|[obama, usa, cina...|
| [currar]|
|[FreePalestine, I...|
| [Taurus]|
| [JangKeunSuk]|
| [thefive]|
| [FF]|
+--------------------+
only showing top 20 rows

h: org.apache.spark.sql.Dataset[Array[String]] = [text: array<string>]
//se va a crear un cuenta hashtags para evaluar que hashtags se repiten más.
//se crea un flatMap donde se mapea los elementos para crear un par clave-valor
val pairs = h.flatMap(p => p.map(n => (n,1)))
pairs.show()
+------------------+---+
| _1| _2|
+------------------+---+
| nra| 1|
| tcot| 1|
| WAR| 1|
| gop| 1|
| teaparty| 1|
| l| 1|
|reasonsobamaislate| 1|
| bias| 1|
| Nra| 1|
| Israel| 1|
| Gaza| 1|
| gun| 1|
| PresidentObama| 1|
| Pakistan| 1|
| GAY| 1|
| NYC| 1|
| Syria| 1|
| FiscalCliff| 1|
| MC| 1|
| Dems| 1|
+------------------+---+
only showing top 20 rows

pairs: org.apache.spark.sql.Dataset[(String, Int)] = [_1: string, _2: int]
//convertir el dataset en dataframe, asignarle nombre a las columnas y agrupar por su Key
val cuenta = pairs.toDF("key","value").groupBy("key").agg(sum("value"))
cuenta.show()
+----------------+----------+
| key|sum(value)|
+----------------+----------+
| igiveup| 1|
| Drink| 1|
| jewelry| 66|
| art| 34|
| Videogames| 2|
| outfit| 12|
| LT| 23|
| Virgo| 486|
| HRW_AR| 1|
| smtlive| 10|
| hope| 10|
| CelticsEspaña| 1|
|VisitSouthAfrica| 1|
| YaMeCanse| 1|
| Tyler| 6|
| anime| 4|
| indigenous| 2|
| NBC| 18|
| travel| 36|
| online| 14|
+----------------+----------+
only showing top 20 rows

cuenta: org.apache.spark.sql.DataFrame = [key: string, sum(value): bigint]
//se ordenan los hashtags con orderBy()
//para ordenar se ha de importar functions
import org.apache.spark.sql.functions._
val cuentaOrdenada = cuenta.orderBy(desc("sum(value)"))
cuentaOrdenada.show()
+---------------+----------+
| key|sum(value)|
+---------------+----------+
| tcot| 2912|
| Pakistan| 2802|
| Obama| 1974|
| NDAA| 1843|
| 23Dec| 1693|
| NRA| 1538|
| black| 1241|
| TuitUtil| 1172|
| teaparty| 1115|
| p2| 1078|
| FiscalCliff| 887|
| FF| 858|
| Libra| 798|
| ChangePK| 662|
| Imagine| 636|
| DrTahirulQadri| 636|
|BelieveAcoustic| 617|
| Capricorn| 613|
| news| 605|
| 1DNews| 549|
+---------------+----------+
only showing top 20 rows

import org.apache.spark.sql.functions._
cuentaOrdenada: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [key: string, sum(value): bigint]

Answer

true

Question 57

Haciendo uso del fichero twitter. * ¿Cuantos tweets contiene el fichero? * ¿Cuantos en inglés? * ¿De cuantos usuarios distintos hay algún tweet en el fichero?

Accepted Answer

Haciendo uso del fichero twitter. * ¿Cuantos tweets contiene el fichero? * ¿Cuantos en inglés? * ¿De cuantos usuarios distintos hay algún tweet en el fichero?

val tweetsJson = spark.read.json("./documentos/large-17-json")
tweetsJson: org.apache.spark.sql.DataFrame = [contributors: string, coordinates: struct<coordinates: array<double>, type: string> ... 22 more fields]
// ¿Cuantos tweets contiene el fichero?
val count = tweetsJson.count()
println("Contiene" + count + " tweets")
Contiene515361 tweets

count: Long = 515361
// Cuantos en inglés?
tweetsJson.printSchema

//se aprecia que el idioma es un String en lang
val tweetsEn = tweetsJson.filter($"lang" === "en")
println("La cantidad de tweets en ingles es: " + tweetsEn.count())
root
|-- contributors: string (nullable = true)
|-- coordinates: struct (nullable = true)
| |-- coordinates: array (nullable = true)
| | |-- element: double (containsNull = true)
| |-- type: string (nullable = true)
|-- created_at: string (nullable = true)
|-- entities: struct (nullable = true)
| |-- hashtags: array (nullable = true)
| | |-- element: struct (containsNull = true)
| | | |-- indices: array (nullable = true)
| | | | |-- element: long (containsNull = true)
| | | |-- text: string (nullable = true)
| |-- media: array (nullable = true)
| | |-- element: struct (containsNull = true)
| | | |-- display_url: string (nullable = true)
| | | |-- expanded_url: string (nullable = true)
| | | |-- id: long (nullable = true)
| | | |-- id_str: string (nullable = true)
| | | |-- indices: array (nullable = true)
| | | | |-- element: long (containsNull = true)
| | | |-- media_url: string (nullable = true)
| | | |-- media_url_https: string (nullable = true)
| | | |-- sizes: struct (nullable = true)
| | | | |-- large: struct (nullable = true)
| | | | | |-- h: long (nullable = true)
| | | | | |-- resize: string (nullable = true)
| | | | | |-- w: long (nullable = true)
| | | | |-- medium: struct (nullable = true)
| | | | | |-- h: long (nullable = true)
| | | | | |-- resize: string (nullable = true)
| | | | | |-- w: long (nullable = true)
| | | | |-- small: struct (nullable = true)
| | | | | |-- h: long (nullable = true)
| | | | | |-- resize: string (nullable = true)
| | | | | |-- w: long (nullable = true)
| | | | |-- thumb: struct (nullable = true)
| | | | | |-- h: long (nullable = true)
| | | | | |-- resize: string (nullable = true)
| | | | | |-- w: long (nullable = true)
| | | |-- source_status_id: long (nullable = true)
| | | |-- source_status_id_str: string (nullable = true)
| | | |-- type: string (nullable = true)
| | | |-- url: string (nullable = true)
| |-- urls: array (nullable = true)
| | |-- element: struct (containsNull = true)
| | | |-- display_url: string (nullable = true)
| | | |-- expanded_url: string (nullable = true)
| | | |-- indices: array (nullable = true)
| | | | |-- element: long (containsNull = true)
| | | |-- url: string (nullable = true)
| |-- user_mentions: array (nullable = true)
| | |-- element: struct (containsNull = true)
| | | |-- id: long (nullable = true)
| | | |-- id_str: string (nullable = true)
| | | |-- indices: array (nullable = true)
| | | | |-- element: long (containsNull = true)
| | | |-- name: string (nullable = true)
| | | |-- screen_name: string (nullable = true)
|-- favorited: boolean (nullable = true)
|-- geo: struct (nullable = true)
| |-- coordinates: array (nullable = true)
| | |-- element: double (containsNull = true)
| |-- type: string (nullable = true)
|-- id: long (nullable = true)
|-- id_str: string (nullable = true)
|-- in_reply_to_screen_name: string (nullable = true)
|-- in_reply_to_status_id: long (nullable = true)
|-- in_reply_to_status_id_str: string (nullable = true)
|-- in_reply_to_user_id: long (nullable = true)
|-- in_reply_to_user_id_str: string (nullable = true)
|-- lang: string (nullable = true)
|-- limit: struct (nullable = true)
| |-- track: long (nullable = true)
|-- place: struct (nullable = true)
| |-- attributes: struct (nullable = true)
| | |-- locality: string (nullable = true)
| | |-- region: string (nullable = true)
| | |-- street_address: string (nullable = true)
| |-- bounding_box: struct (nullable = true)
| | |-- coordinates: array (nullable = true)
| | | |-- element: array (containsNull = true)
| | | | |-- element: array (containsNull = true)
| | | | | |-- element: double (containsNull = true)
| | |-- type: string (nullable = true)
| |-- country: string (nullable = true)
| |-- country_code: string (nullable = true)
| |-- full_name: string (nullable = true)
| |-- id: string (nullable = true)
| |-- name: string (nullable = true)
| |-- place_type: string (nullable = true)
| |-- url: string (nullable = true)
|-- possibly_sensitive: boolean (nullable = true)
|-- retweet_count: long (nullable = true)
|-- retweeted: boolean (nullable = true)
|-- retweeted_status: struct (nullable = true)
| |-- contributors: array (nullable = true)
| | |-- element: long (containsNull = true)
| |-- coordinates: struct (nullable = true)
| | |-- coordinates: array (nullable = true)
| | | |-- element: double (containsNull = true)
| | |-- type: string (nullable = true)
| |-- created_at: string (nullable = true)
| |-- entities: struct (nullable = true)
| | |-- hashtags: array (nullable = true)
| | | |-- element: struct (containsNull = true)
| | | | |-- indices: array (nullable = true)
| | | | | |-- element: long (containsNull = true)
| | | | |-- text: string (nullable = true)
| | |-- media: array (nullable = true)
| | | |-- element: struct (containsNull = true)
| | | | |-- display_url: string (nullable = true)
| | | | |-- expanded_url: string (nullable = true)
| | | | |-- id: long (nullable = true)
| | | | |-- id_str: string (nullable = true)
| | | | |-- indices: array (nullable = true)
| | | | | |-- element: long (containsNull = true)
| | | | |-- media_url: string (nullable = true)
| | | | |-- media_url_https: string (nullable = true)
| | | | |-- sizes: struct (nullable = true)
| | | | | |-- large: struct (nullable = true)
| | | | | | |-- h: long (nullable = true)
| | | | | | |-- resize: string (nullable = true)
| | | | | | |-- w: long (nullable = true)
| | | | | |-- medium: struct (nullable = true)
| | | | | | |-- h: long (nullable = true)
| | | | | | |-- resize: string (nullable = true)
| | | | | | |-- w: long (nullable = true)
| | | | | |-- small: struct (nullable = true)
| | | | | | |-- h: long (nullable = true)
| | | | | | |-- resize: string (nullable = true)
| | | | | | |-- w: long (nullable = true)
| | | | | |-- thumb: struct (nullable = true)
| | | | | | |-- h: long (nullable = true)
| | | | | | |-- resize: string (nullable = true)
| | | | | | |-- w: long (nullable = true)
| | | | |-- source_status_id: long (nullable = true)
| | | | |-- source_status_id_str: string (nullable = true)
| | | | |-- type: string (nullable = true)
| | | | |-- url: string (nullable = true)
| | |-- urls: array (nullable = true)
| | | |-- element: struct (containsNull = true)
| | | | |-- display_url: string (nullable = true)
| | | | |-- expanded_url: string (nullable = true)
| | | | |-- indices: array (nullable = true)
| | | | | |-- element: long (containsNull = true)
| | | | |-- url: string (nullable = true)
| | |-- user_mentions: array (nullable = true)
| | | |-- element: struct (containsNull = true)
| | | | |-- id: long (nullable = true)
| | | | |-- id_str: string (nullable = true)
| | | | |-- indices: array (nullable = true)
| | | | | |-- element: long (containsNull = true)
| | | | |-- name: string (nullable = true)
| | | | |-- screen_name: string (nullable = true)
| |-- favorited: boolean (nullable = true)
| |-- geo: struct (nullable = true)
| | |-- coordinates: array (nullable = true)
| | | |-- element: double (containsNull = true)
| | |-- type: string (nullable = true)
| |-- id: long (nullable = true)
| |-- id_str: string (nullable = true)
| |-- in_reply_to_screen_name: string (nullable = true)
| |-- in_reply_to_status_id: long (nullable = true)
| |-- in_reply_to_status_id_str: string (nullable = true)
| |-- in_reply_to_user_id: long (nullable = true)
| |-- in_reply_to_user_id_str: string (nullable = true)
| |-- lang: string (nullable = true)
| |-- place: struct (nullable = true)
| | |-- attributes: struct (nullable = true)
| | | |-- street_address: string (nullable = true)
| | |-- bounding_box: struct (nullable = true)
| | | |-- coordinates: array (nullable = true)
| | | | |-- element: array (containsNull = true)
| | | | | |-- element: array (containsNull = true)
| | | | | | |-- element: double (containsNull = true)
| | | |-- type: string (nullable = true)
| | |-- country: string (nullable = true)
| | |-- country_code: string (nullable = true)
| | |-- full_name: string (nullable = true)
| | |-- id: string (nullable = true)
| | |-- name: string (nullable = true)
| | |-- place_type: string (nullable = true)
| | |-- url: string (nullable = true)
| |-- possibly_sensitive: boolean (nullable = true)
| |-- retweet_count: long (nullable = true)
| |-- retweeted: boolean (nullable = true)
| |-- scopes: struct (nullable = true)
| | |-- followers: boolean (nullable = true)
| |-- source: string (nullable = true)
| |-- text: string (nullable = true)
| |-- truncated: boolean (nullable = true)
| |-- user: struct (nullable = true)
| | |-- contributors_enabled: boolean (nullable = true)
| | |-- created_at: string (nullable = true)
| | |-- default_profile: boolean (nullable = true)
| | |-- default_profile_image: boolean (nullable = true)
| | |-- description: string (nullable = true)
| | |-- favourites_count: long (nullable = true)
| | |-- follow_request_sent: string (nullable = true)
| | |-- followers_count: long (nullable = true)
| | |-- following: string (nullable = true)
| | |-- friends_count: long (nullable = true)
| | |-- geo_enabled: boolean (nullable = true)
| | |-- id: long (nullable = true)
| | |-- id_str: string (nullable = true)
| | |-- is_translator: boolean (nullable = true)
| | |-- lang: string (nullable = true)
| | |-- listed_count: long (nullable = true)
| | |-- location: string (nullable = true)
| | |-- name: string (nullable = true)
| | |-- notifications: string (nullable = true)
| | |-- profile_background_color: string (nullable = true)
| | |-- profile_background_image_url: string (nullable = true)
| | |-- profile_background_image_url_https: string (nullable = true)
| | |-- profile_background_tile: boolean (nullable = true)
| | |-- profile_banner_url: string (nullable = true)
| | |-- profile_image_url: string (nullable = true)
| | |-- profile_image_url_https: string (nullable = true)
| | |-- profile_link_color: string (nullable = true)
| | |-- profile_sidebar_border_color: string (nullable = true)
| | |-- profile_sidebar_fill_color: string (nullable = true)
| | |-- profile_text_color: string (nullable = true)
| | |-- profile_use_background_image: boolean (nullable = true)
| | |-- protected: boolean (nullable = true)
| | |-- screen_name: string (nullable = true)
| | |-- statuses_count: long (nullable = true)
| | |-- time_zone: string (nullable = true)
| | |-- url: string (nullable = true)
| | |-- utc_offset: long (nullable = true)
| | |-- verified: boolean (nullable = true)
|-- source: string (nullable = true)
|-- text: string (nullable = true)
|-- truncated: boolean (nullable = true)
|-- user: struct (nullable = true)
| |-- contributors_enabled: boolean (nullable = true)
| |-- created_at: string (nullable = true)
| |-- default_profile: boolean (nullable = true)
| |-- default_profile_image: boolean (nullable = true)
| |-- description: string (nullable = true)
| |-- favourites_count: long (nullable = true)
| |-- follow_request_sent: string (nullable = true)
| |-- followers_count: long (nullable = true)
| |-- following: string (nullable = true)
| |-- friends_count: long (nullable = true)
| |-- geo_enabled: boolean (nullable = true)
| |-- id: long (nullable = true)
| |-- id_str: string (nullable = true)
| |-- is_translator: boolean (nullable = true)
| |-- lang: string (nullable = true)
| |-- listed_count: long (nullable = true)
| |-- location: string (nullable = true)
| |-- name: string (nullable = true)
| |-- notifications: string (nullable = true)
| |-- profile_background_color: string (nullable = true)
| |-- profile_background_image_url: string (nullable = true)
| |-- profile_background_image_url_https: string (nullable = true)
| |-- profile_background_tile: boolean (nullable = true)
| |-- profile_banner_url: string (nullable = true)
| |-- profile_image_url: string (nullable = true)
| |-- profile_image_url_https: string (nullable = true)
| |-- profile_link_color: string (nullable = true)
| |-- profile_sidebar_border_color: string (nullable = true)
| |-- profile_sidebar_fill_color: string (nullable = true)
| |-- profile_text_color: string (nullable = true)
| |-- profile_use_background_image: boolean (nullable = true)
| |-- protected: boolean (nullable = true)
| |-- screen_name: string (nullable = true)
| |-- statuses_count: long (nullable = true)
| |-- time_zone: string (nullable = true)
| |-- url: string (nullable = true)
| |-- utc_offset: long (nullable = true)
| |-- verified: boolean (nullable = true)

La cantidad de tweets en ingles es: 356218

tweetsEn: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [contributors: string, coordinates: struct<coordinates: array<double>, type: string> ... 22 more fields]
// ¿De cuantos usuarios distintos hay algún tweet en el fichero?
val usuarios = tweetsJson.select("user.screen_name")
val distinct = usuarios.distinct().count()
usuarios: org.apache.spark.sql.DataFrame = [screen_name: string]
distinct: Long = 352736

Answer

true

Question 58

Define una clase tweet con la siguiente estructura: - Tweet - Fecha creacion - Nombre usuario - Numero seguidores - Texto

Luego, convierte el DataFrame del Ejercicio 1 en un DataSet.
¿Quien es el usuario con mayor número de seguidores?

Accepted Answer

// vamos a trabajar con datasets
import spark.implicits._
import spark.implicits._
// defenir clase Tweet
case class Tweet(fecha_creacion: Option[String], usuario: Option[String], 
 numSeguidores: Option[Long], texto: Option[String])
defined class Tweet
// selección de las columnas
val tweetsDF = tweetsJson.select("created_at", "user.screen_name", "user.followers_count", "text")

tweetOrdenado.select("usuario").limit(1).show()
+-----------+
| usuario|
+-----------+
|BarackObama|
+-----------+

tweetOrdenado: org.apache.spark.sql.Dataset[Tweet] = [fecha_creacion: string, usuario: string ... 2 more fields]

Answer

true

Question 59

Haciendo uso del Dataset del ejercicio anterior - Aplícale persistencia sólo en disco. - ¿Cuántos tweets contienen la palabra “data”?. - Volcar el dataset en un fichero JSON.

Accepted Answer

dataDS: org.apache.spark.sql.Dataset[Array[String]] = [value: array<string>]
// se filtran los que contengan la palabra "data"
dataDS.filter(_.contains("data")).count()
res34: Long = 161
// volcar el Dataset en un fichero Json
dataDS.write.json("./documentos/datasetTweet")

Answer

true

Question 60

Crear una nueva sesión de spark y cargar los datos del fichero “large-17-json”.
Crear un dataframe con los 50 tweets más ‘retweeteados’ ordenados por ese campo. Sólo nos interesan los campos retweeted_status.user.screen_name, retweeted_status.text, retweeted_status.retweet_count. No pueden aparecer tweets con número de RT = 0. Por último, vuelca estos datos en un fichero JSON.

Accepted Answer

Answer

true

Question 61

Crear un programa el cual, recibiendo ficheros con la información acerca del nombre, área y duración de cursos, muestre la duración de cada área.

Accepted Answer

//como se va a trabajar con ficheros se realizan las importaciones
import spark.implicits._
import org.apache.spark.sql.types._

val schema = new StructType().add("nombre","string").add("area","string").add("duracion","long")
import spark.implicits._
import org.apache.spark.sql.types._
schema: org.apache.spark.sql.types.StructType = StructType(StructField(nombre,StringType,true), StructField(area,StringType,true), StructField(duracion,LongType,true))
//se leen los ficheros
val cursosDF = spark.readStream
 .option("sep", ";")
 .schema(schema)
 .csv("./documentos/cursos/")

//se agrupa por el campo área y se selecciona el curso con la máxima duración de cada área
val cursosDuracionMayor = cursosDF.groupBy($"area").agg(max("duracion"))

val query = cursosDuracionMayor.writeStream
 .format("console")
 .outputMode("complete")
 .start()

query.awaitTermination(90000)
-------------------------------------------
Batch: 0
-------------------------------------------
+----+-------------+
|area|max(duracion)|
+----+-------------+
+----+-------------+

-------------------------------------------
Batch: 1
-------------------------------------------
+------------+-------------+
| area|max(duracion)|
+------------+-------------+
| Estadistica| 9|
|Programacion| 8|
| BIGDATA| 10|
+------------+-------------+

cursosDF: org.apache.spark.sql.DataFrame = [nombre: string, area: string ... 1 more field]
cursosDuracionMayor: org.apache.spark.sql.DataFrame = [area: string, max(duracion): bigint]
query: org.apache.spark.sql.streaming.StreamingQuery = org.apache.spark.sql.execution.streaming.StreamingQueryWrapper@7879fba2
res3: Boolean = false

Answer

true

Question 62

- Streaming
Crear un programa que, recibiendo en tiempo real los ficheros adjuntos con nombre “ej2_XXX.json”, muestre por pantalla el número de tweets escritos por cada usuario de mayor a menor. Los fichero son fecha de creación, nombre de usuario y texto.

Accepted Answer

//como se va a trabajar con ficheros se realizan las importaciones
import spark.implicits._
import org.apache.spark.sql.types._
import java.sql.Timestamp
import org.apache.spark.sql.functions._

val schema = new StructType().add("created_at",TimestampType).add("screen_name","string").add("text","string")

val tweetsDF = spark.readStream
 .schema(schema)
 .json("./documentos/spanishTweets")

val numTweets = tweetsDF.groupBy($"screen_name").count().orderBy(desc("count"))

val query = numTweets.writeStream
 .format("console")
 .outputMode("complete")
 .start()

query.awaitTermination(90000)
-------------------------------------------
Batch: 0
-------------------------------------------
+---------------+-----+
| screen_name|count|
+---------------+-----+
| ___Cinthya___| 41|
| valepalacio| 36|
| rockmeezayn| 29|
| carmencel95| 28|
|CucharasMyWorld| 25|
| PauBarajas17| 24|
| OhLaLaCarrots| 23|
|LatinNiallLover| 23|
| CoverGirls_BTR| 23|
| Nayaretttth| 22|
| Blaaasi| 21|
|JavierArgueta19| 21|
| staystro0ng| 20|
| Phrases_Buenas| 20|
| henderwhore88| 20|
| QueridoCerdito| 20|
|_Latinamerica1D| 18|
| Harrytecojo| 18|
| OdairsSea| 18|
| luciatisera| 18|
+---------------+-----+
only showing top 20 rows

import spark.implicits._
import org.apache.spark.sql.types._
import java.sql.Timestamp
import org.apache.spark.sql.functions._
schema: org.apache.spark.sql.types.StructType = StructType(StructField(created_at,TimestampType,true), StructField(screen_name,StringType,true), StructField(text,StringType,true))
tweetsDF: org.apache.spark.sql.DataFrame = [created_at: timestamp, screen_name: string ... 1 more field]
numTweets: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [screen_name: string, count: bigint]
query: org.apache.spark.sql.streaming.StreamingQuery = org.apache.spark.sql.execution.streaming.StreamingQueryWrapper@1fa57e46
res5: Boolean = false

Answer

true

Question 63

Join
Dado un dataset que ya tenemos (estático), se leerá un nuevo dataset y se creará un Join en streaming. El dataset contendrá las variables nombre, puesto y salario.

Accepted Answer

import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

import spark.implicits._

//se crea la estructura que contrendrá los datos leidos.
val schema = new StructType().add("nombre", "string").add("puesto","string").add("salario","long")

//se leen todos los ficheros
val staticPersonas = spark.read.json("./documentos/personas/*")
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._
import spark.implicits._
schema: org.apache.spark.sql.types.StructType = StructType(StructField(nombre,StringType,true), StructField(puesto,StringType,true), StructField(salario,LongType,true))
staticPersonas: org.apache.spark.sql.DataFrame = [edad: bigint, nombre: string ... 1 more field]
//ficheros que se leen en streaming
val streamingSalarios = spark.readStream
.schema(schema)
.json("./documentos/personas_salarios")

//unión de los ficheros. Se indican los ficheros y el campo por el que se quiere realizar el join.
val both = streamingSalarios.join(staticPersonas, "nombre")

//ahora se creará una vista y con SQL se calculará la media de salario por puesto.
both.createOrReplaceTempView("personas_salario")

val mediaSalario = spark.sql("SELECT puesto, avg(salario) FROM personas_salario GROUP BY puesto")

val query = mediaSalario.writeStream
.outputMode("complete")
.format("console")
.start()

query.awaitTermination(90000)
-------------------------------------------
Batch: 0
-------------------------------------------
+-------------+------------+
| puesto|avg(salario)|
+-------------+------------+
|Desarrollador| 11.0|
| Analista| 15.0|
+-------------+------------+

-------------------------------------------
Batch: 1
-------------------------------------------
+-------------+------------+
| puesto|avg(salario)|
+-------------+------------+
|Desarrollador| 12.0|
| Analista| 17.0|
+-------------+------------+

-------------------------------------------
Batch: 2
-------------------------------------------
+-------------+------------------+
| puesto| avg(salario)|
+-------------+------------------+
| JefeProyecto| 28.0|
|Desarrollador|12.714285714285714|
| Analista| 17.0|
+-------------+------------------+

streamingSalarios: org.apache.spark.sql.DataFrame = [nombre: string, puesto: string ... 1 more field]
both: org.apache.spark.sql.DataFrame = [nombre: string, puesto: string ... 3 more fields]
mediaSalario: org.apache.spark.sql.DataFrame = [puesto: string, avg(salario): double]
query: org.apache.spark.sql.streaming.StreamingQuery = org.apache.spark.sql.execution.streaming.StreamingQueryWrapper@3d71130c
res8: Boolean = false
//para obtener el id de la query
query.id
res9: java.util.UUID = 5871ddc1-2dd8-41bb-8de5-dcfa98d153ef
//información de la query
query.explain()
== Physical Plan ==
WriteToDataSourceV2 org.apache.spark.sql.execution.streaming.sources.MicroBatchWriter@430dc3d6
+- *(5) HashAggregate(keys=[puesto#5566], functions=[avg(salario#5567L)])
+- StateStoreSave [puesto#5566], state info [ checkpoint = file:/tmp/temporary-78cf70d4-3979-4c2a-baad-6cfe6772a40f/state, runId = 6253af8e-6f52-416b-99e4-8783ac1f5274, opId = 0, ver = 2, numPartitions = 200], Complete, 0
+- *(4) HashAggregate(keys=[puesto#5566], functions=[merge_avg(salario#5567L)])
+- StateStoreRestore [puesto#5566], state info [ checkpoint = file:/tmp/temporary-78cf70d4-3979-4c2a-baad-6cfe6772a40f/state, runId = 6253af8e-6f52-416b-99e4-8783ac1f5274, opId = 0, ver = 2, numPartitions = 200]
+- *(3) HashAggregate(keys=[puesto#5566], functions=[merge_avg(salario#5567L)])
+- Exchange hashpartitioning(puesto#5566, 200)
+- *(2) HashAggregate(keys=[puesto#5566], functions=[partial_avg(salario#5567L)])
+- *(2) Project [puesto#5566, salario#5567L]
+- *(2) BroadcastHashJoin [nombre#5565], [nombre#2434], Inner, BuildRight
:- *(2) Project [nombre#5565, puesto#5566, salario#5567L]
: +- *(2) Filter isnotnull(nombre#5565)
: +- *(2) FileScan json [nombre#5565,puesto#5566,salario#5567L] Batched: false, Format: JSON, Location: InMemoryFileIndex[file:/vagrant/IPNB/Ejercicios/documentos/personas_salarios/salario3, file:/vagr..., PartitionFilters: [], PushedFilters: [IsNotNull(nombre)], ReadSchema: struct<nombre:string,puesto:string,salario:bigint>
+- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]))
+- *(1) Project [nombre#2434]
+- *(1) Filter isnotnull(nombre#2434)
+- *(1) FileScan json [nombre#2434] Batched: false, Format: JSON, Location: InMemoryFileIndex[file:/vagrant/IPNB/Ejercicios/documentos/personas/nombres1, file:/vagrant/IPNB/..., PartitionFilters: [], PushedFilters: [IsNotNull(nombre)], ReadSchema: struct<nombre:string>
//name de la query (no se le ha dado ninguno)
query.name
res11: String = null
//más información
query.lastProgress
res12: org.apache.spark.sql.streaming.StreamingQueryProgress =
{
"id" : "5871ddc1-2dd8-41bb-8de5-dcfa98d153ef",
"runId" : "6253af8e-6f52-416b-99e4-8783ac1f5274",
"name" : null,
"timestamp" : "2018-07-24T17:27:43.676Z",
"batchId" : 3,
"numInputRows" : 0,
"inputRowsPerSecond" : 0.0,
"processedRowsPerSecond" : 0.0,
"durationMs" : {
"getOffset" : 356,
"triggerExecution" : 356
},
"stateOperators" : [ {
"numRowsTotal" : 3,
"numRowsUpdated" : 0,
"memoryUsedBytes" : 18055
} ],
"sources" : [ {
"description" : "FileStreamSource[file:/vagrant/IPNB/Ejercicios/documentos/personas_salarios]",
"startOffset" : {
"logOffset" : 2
},
"endOffset" : {
"logOffset" : 2
},
"numInputRows" : 0,
"inputRowsPerSecond" : 0.0,
"...
//Comprobar si está activa la query
query.isActive
res13: Boolean = true
//para la query
query.stop

Answer

true

Question 64

Ventanas Streaming
Crear RDD para ventana de datos de 10 segundos.

Accepted Answer

//se realizan las importaciones para manejo de DStream.
import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext

val conf = new SparkConf().setMaster("local[4]").setAppName("miApp")
 .set("spark.driver.allowMultipleContexts", "true")

//se crea el StreamingContext con la configuración y ejecución cada 5 segundos.
val ssc = new StreamingContext(conf, Seconds(5))

//se prepara para leer del socket
val lines = ssc.socketTextStream("localhost", 9999)
val linesWindows = lines.window(Seconds(10))

//se crea un cuenta palabras leyendo desde el socket
val palabras = linesWindows.flatMap(_.split(" "))

val pairs = palabras.map(w => (w,1))
val cuenta = pairs.reduceByKey((a,b) => a+b)

cuenta.print()

//start para comenzar
ssc.start()
ssc.awaitTerminationOrTimeout(45000)

Answer

true

Ohne Titel

Erstelle ein kostenloses Konto, um dieses Quiz zu beenden

Fragen in diesem Quiz (82)

Informationen

Feedback

Wie gefällt dir das neue Design?