Test
Sin título
Crea una cuenta gratis para terminar este test
Guarda tu progreso y consulta tu resultado final.
¿Ya tienes cuenta? Inicia sesión
Preguntas del test (82)
1¿Qué es Apache Spark?
- AApache 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. - BTrue
- AApache Spark es un sistema de computación distribuido de código abierto basado en Hadoop.
2¿Cuáles son las características más notables de Apache Spark?
- Avelocidad, 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. - Btrue
- Avelocidad, compatibilidad multi-formato y librerías integradas.
3¿Qué es "SCC"?
- ASCC, 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.
- Btrue
4¿Sabes qué es Apache Hadoop
- AHadoop 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. - Btrue
- AHadoop nace como iniciativa de Apache para dar soporte al paradigma de programación Map-Reduce, que fue inicialmente publicado por Google.
5¿Qué es MapReduce?
- AMapReduce 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.
- Btrue
6¿Qué es "RDD"?
- AResilient 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.
- Btrue
7¿Qué es "inmutabilidad"?
- Acuando un elemento es inmutable, no puede ser cambiado o alterado de ninguna forma una vez se ha creado y asignado un valor.
- Btrue
8¿Qué es YARN?
- AYARN, 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.
- Btrue
9¿Cuál es el lenguaje de programación más utilizado en Spark?
- AAunque existen muchos desarrolladores que les gusta utilizar Python, Scala sigue siendo el lenguaje de programación más utilizado para Apache Spark.
- Btrue
10¿Cuántos administradores de clústeres se encuentran disponibles en Spark?
- A3 , YARN, Apache Mesos y despliegues independientes.
- Btrue
11¿Cuáles son las responsabilidades del motor Spark?
- AGeneralmente, 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.
- Btrue
12Pregunta 10: ¿Qué son las "evaluaciones perezosas"?
- Aeste 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.
- Btrue
13 ¿Qué son las "particiones"?
- AUna 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.
- Btrue
14 ¿Qué es la integración continua?
- Ana 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.
- Btrue
15Qué es Jenkins?
- AJenkins 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. - Btrue
- AJenkins 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.
16Que es SonarQube
- AEs 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) - Btrue
- AEs 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.
17¿Para qué se utiliza el Apache Spark Streaming?
- ASpark 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.
- Btrue
18¿Es normal ejecutar todos tus procesos en un nodo localizado?
- ANo, 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.
- Btrue
19¿Para qué se utiliza "SparkCore"?
- ASparkCore, 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.
- Btrue
20 ¿La API del sistema de archivos tiene un uso en Apache Spark?
- AClaro que si, Esta API en particular le permite a Spark leer y crear los datos a través de diferentes áreas de almacenamiento (Dispositivos).
- Btrue
211. Apache Spark Core
- ATodas 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.
- Btrue
222. Apache Spark SQL
- AEl 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.
- Btrue
233. Apache Spark Streaming
- AEs 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 . - Btrue
- AEs 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.
24¿Cómo funciona Apache Spark?
- AFunciona 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. - Btrue
- AFunciona 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.
25Apache Spark tiene los siguientes componentes:
- ASpark 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 - Btrue
- ASpark 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.
26 ¿Hay varios patrones de implementación para Spark, cada modo funciones?
- ALocal: 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. - Btrue
- ALocal: comúnmente utilizado para las pruebas de desarrollo local, local también se divide en un multithreading local de un solo roscado y local;
27¿Qué niveles de almacenamiento tienen Spark?
- A1) 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. - Btrue
- A1) Memory_only: los datos se guardan en la memoria, y si la memoria no es suficiente, los datos pueden no durar.
28¿Qué es DataFrame?
- A¿Qué es DataFrame?
- BDataFrame es un conjunto de datos distribuidos basado en RDD, similar a las tablas bidimensionales de las bases de datos tradicionales.
29¿Cuál es la principal diferencia entre DataFrame y RDD?
- AEl 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.
- Btrue
30Funciones de DataFrame
- AFunciones de DataFrame
- B2. 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
31 Conceptos básicos de RDD
- ACliente: 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. - Btrue
- ACliente: El proceso del cliente es responsable de enviar trabajos al Máster.
32 Términos comunes de RDD
- ADAGScheduler: 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. - Btrue
- ADAGScheduler: 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.
33RDD proporciona dos tipos de operaciones
- Atransformació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 - Btrue
- Atransformación y acción
34La diferencia entre transformación y acción en RDD
- ALa 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.
- Btrue
35¿Cuál es la mayor diferencia entre RDD y DSM?
- ADSM(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. - Btrue
- ADSM(distributed shared memory)
36 ¿Cuáles son las ventajas de RDD?
- A1. 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 - Btrue
- A1. Mecanismo eficiente de tolerancia a fallas
379. ¿Cómo obtener RDD?
- A1. 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. - Btrue
- A1. Obtenga del sistema de archivos compartidos (como HDFS)
38Dos tipos de RDD que dependerán de
- ADependencias 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; - Btrue
- ADependencias estrechas (dependencias estrechas) y dependencias amplias (dependencias amplias).
39Principios básicos de Spark Streaming
- AEl 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
- Btrue
40
Case 1Potencia una lista y salida- Aval input = sc.parallelize(List(1,2,3,4))
val result = input.map(x => x*x)
println(result.collect().mkString(",")) - Btrue
- Aval input = sc.parallelize(List(1,2,3,4))
41
Case 3 Imprimir elementos rdd- Ardd.foreach (println) o rdd.map (println).
rdd.collect().foreach(println）
rdd.take(100).foreach(println） - Btrue
- Ardd.foreach (println) o rdd.map (println).
42
Case 2 wordcount- Aval 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)) - Btrue
- Aval textFile = sc.textFile(args(1))
43Que es having
- ALa 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 - Btrue
- ALa función HAVING se utiliza para incluir condiciones con alguna función SQL del tipo SUM, MAX, ..
44Diferencia entre Groupby, distinc y having
- AUtilice DISTINCT si solo desea eliminar duplicados. Utilice GROUPY BY si desea aplicar operadores agregados (MAX, SUM, GROUP_CONCAT, ..., o una cláusula HAVING).
- Btrue
45Que es 1.Flume
- AFlume 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. - Btrue
- AFlume es un sistema de recolección de registros distribuido
46¿Qué es un motor de búsqueda?
- AUn 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.
- Btrue
47¿Qué es Lucene?
- ALucene 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.
- Btrue
48¿Qué es Elasticsearch?
- AElasticsearch 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. - Btrue
- AElasticsearch es una herramienta de análisis y búsqueda de texto completo de código abierto altamente escalable.
49¿Qué es Hive?
- AHive 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
- Btrue
50Importar librerias en scala
- Aimport org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD - Btrue
- Aimport org.apache.spark.SparkConf
51Crea 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.- Aval 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 - Btrue
- Aval array = Array(23,54,32,66,0,-14,-8,10)
52Haciendo 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.
- A//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)) - Btrue
- A//carga del fichero loremipsum.txt
53Realiza 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”
- Aval 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 - Btrue
- Aval mayor6 = lorem.flatMap(line => line.split(" ")).filter(p => p.length() >= 6)
54Dataframe y Datasets
- Aimport org.apache.spark.sql.Dataset
import org.apache.spark.sql.DataFrame
import spark.implicits._
import org.apache.spark.sql.Dataset
import org.apache.spark.sql.DataFrame
import spark.implicits._
//carga de fichero json
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]
//imprimir esquema
tweetsJson.printSchema
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)
//textos de los tweets, seleccionar los tres primeros.
tweetsJson.select("text").head(3)
res4: Array[org.apache.spark.sql.Row] = Array([Obama meets Reid on fiscal cliff, plans remarks http://t.co/Nuec5Wh6], [@halluzaynating obama], [@hii1d Awww. Poor Niall. :(( :)])
//como tweetsJson tiene muchas variables se va a crear un dataset a partir del dataframe sólo con algunas variables.
case class Tweet(fecha_creacion: Option[String], usuario: Option[String],
isRT: Option[Boolean], numRT: Option[Long], texto: Option[String],
idioma: Option[String], tweet_sensible: Boolean, source: String,
favorito: Option[Boolean], numSeguidores: Option[Long])
val tweetsDF = tweetsJson.select("created_at", "user.screen_name", "retweeted", "retweet_count",
"text", "lang", "possibly_sensitive", "source", "retweeted_status.favorited",
"user.followers_count")
tweetsDF.show()
+--------------------+---------------+---------+-------------+--------------------+----+------------------+--------------------+---------+---------------+
| created_at| screen_name|retweeted|retweet_count| text|lang|possibly_sensitive| source|favorited|followers_count|
+--------------------+---------------+---------+-------------+--------------------+----+------------------+--------------------+---------+---------------+
|Fri Dec 21 22:53:...| washexaminer| false| 0|Obama meets Reid ...| en| false|<a href="http://t...| null| 1995|
|Fri Dec 21 22:53:...|narrychristmas_| false| 0|@halluzaynating o...| en| null| web| null| 269|
|Fri Dec 21 22:53:...| nicolehoran08| false| 0|@hii1d Awww. Poor...| pl| null| web| null| 120|
|Fri Dec 21 22:53:...| Odemanjr15| false| 0|All you need in a...| en| null|<a href="http://t...| null| 61|
|Fri Dec 21 22:53:...| colovion| false| 0|RT @PatDollard: W...| en| false|<a href="http://t...| false| 490|
|Fri Dec 21 22:53:...|Coolan_n_Foolan| false| 0|Crankin to this F...| en| null|<a href="http://t...| null| 153|
|Fri Dec 21 22:53:...| YoSoyDalia| false| 0|RT @InjusticeFact...| en| null| web| false| 2081|
|Fri Dec 21 22:53:...| KinderdijkNL| false| 0|Prio 2 TS500 Bran...| nl| false|<a href="http://w...| null| 42|
|Fri Dec 21 22:53:...| akmalarif90| false| 0|RT @BBCWorld: Bar...| en| false|<a href="http://w...| false| 545|
|Fri Dec 21 22:53:...| organicsblue| false| 0|Amazon Kindle Pap...| en| false|<a href="http://w...| null| 1|
|Fri Dec 21 22:53:...| Odielicious18| false| 0|Magna now, West J...| en| null|<a href="http://t...| null| 729|
|Fri Dec 21 22:53:...| ArticLightning| false| 0|I liked a @YouTub...| en| false|<a href="http://w...| null| 168|
|Fri Dec 21 22:53:...| skyriver1| false| 0|If social media w...| en| null|<a href="http://t...| null| 679|
|Fri Dec 21 22:53:...| MAliciavp| false| 0|RT @Clapsico: OTR...| es| false|<a href="http://t...| false| 435|
|Fri Dec 21 22:53:...| dave53elmer| false| 0|RT @meridak Obam...| en| null|<a href="http://w...| null| 127|
|Fri Dec 21 22:53:...| onegringaa| false| 0|RT @OceannJennaa:...| en| null|<a href="http://t...| false| 331|
|Fri Dec 21 22:53:...| FingerStinger| false| 0|"Bred" definition...| en| null|<a href="http://u...| null| 5353|
|Fri Dec 21 22:53:...| mario87s| false| 0|RT @atienzajc: Ll...| es| null|<a href="http://t...| false| 111|
|Fri Dec 21 22:53:...| brownandbella| false| 0|RT @Green_Footbal...| en| null|<a href="http://w...| false| 788|
|Fri Dec 21 22:53:...| zhaiickodrjv| false| 0|He's playin' Call...| en| null|<a href="http://t...| null| 1604|
+--------------------+---------------+---------+-------------+--------------------+----+------------------+--------------------+---------+---------------+
only showing top 20 rows
defined class Tweet
tweetsDF: org.apache.spark.sql.DataFrame = [created_at: string, screen_name: string ... 8 more fields]
val tweetsDS = tweetsDF.toDF("fecha_creacion", "usuario", "isRT", "numRT",
"texto", "idioma", "tweet_sensible", "source",
"favorito", "numSeguidores").as[Tweet]
tweetsDS: org.apache.spark.sql.Dataset[Tweet] = [fecha_creacion: string, usuario: string ... 8 more fields] - Btrue
- Aimport org.apache.spark.sql.Dataset
55Filtrar los tweets en español.
- A//filtrar el texto y el usuario de los tweets en español.
val tweetSpanish = tweetsDS.filter($"lang" === "es").select("texto", "usuario").show()
+--------------------+---------------+
| texto| usuario|
+--------------------+---------------+
|RT @Clapsico: OTR...| MAliciavp|
|RT @atienzajc: Ll...| mario87s|
|@onedirecshon htt...| runforawarzone|
|Que Mi Media Nara...|NicolCardenas11|
|"@globovision: Ob...| greichis|
|No pega que tenga...|juliannerxbedwe|
|Larry aun existe ...| LWWY_LT_1D|
|RT @MiPrincesoMal...| Ladyandboss|
|Si, http://t.co/H...| Tururunes|
|Me piro a sobar q...| elbabas11|
|RT @JulianGottig1...| mati_q1000|
|RT @Anne375: #FF ...| Karnoumz|
|@jossymi28 aww! T...| EXOPlanetSpain|
|Obama está hablan...| FINANCEXXI|
|me ha tocado medi...|Jero5realmadrid|
|Menos buscas más ...|MauriCristiano_|
|1.7 Cubic-Ft Refr...| workkitchens|
|Necesito verla no...| Jesusraxd|
|https://t.co/aNOw...| Betchakian|
|Yo cuando ponen l...| sugardududu|
+--------------------+---------------+
only showing top 20 rows
tweetSpanish: Unit = () - Btrue
- A//filtrar el texto y el usuario de los tweets en español.
56Obtener los hashtags tranding topic
- AObtener 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] - Btrue
- AObtener los hashtags tranding topic
57Haciendo uso del fichero twitter. * ¿Cuantos tweets contiene el fichero? * ¿Cuantos en inglés? * ¿De cuantos usuarios distintos hay algún tweet en el fichero?
- AHaciendo 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 - Btrue
- AHaciendo uso del fichero twitter. * ¿Cuantos tweets contiene el fichero? * ¿Cuantos en inglés? * ¿De cuantos usuarios distintos hay algún tweet en el fichero?
58Define 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?- A// 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")
// conversión a Dataset
val tweetsDS = tweetsDF.toDF("fecha_creacion", "usuario", "numSeguidores", "texto").as[Tweet]
tweetsDF: org.apache.spark.sql.DataFrame = [created_at: string, screen_name: string ... 2 more fields]
tweetsDS: org.apache.spark.sql.Dataset[Tweet] = [fecha_creacion: string, usuario: string ... 2 more fields]
tweetsDS.show()
+--------------------+---------------+-------------+--------------------+
| fecha_creacion| usuario|numSeguidores| texto|
+--------------------+---------------+-------------+--------------------+
|Fri Dec 21 22:53:...| washexaminer| 1995|Obama meets Reid ...|
|Fri Dec 21 22:53:...|narrychristmas_| 269|@halluzaynating o...|
|Fri Dec 21 22:53:...| nicolehoran08| 120|@hii1d Awww. Poor...|
|Fri Dec 21 22:53:...| Odemanjr15| 61|All you need in a...|
|Fri Dec 21 22:53:...| colovion| 490|RT @PatDollard: W...|
|Fri Dec 21 22:53:...|Coolan_n_Foolan| 153|Crankin to this F...|
|Fri Dec 21 22:53:...| YoSoyDalia| 2081|RT @InjusticeFact...|
|Fri Dec 21 22:53:...| KinderdijkNL| 42|Prio 2 TS500 Bran...|
|Fri Dec 21 22:53:...| akmalarif90| 545|RT @BBCWorld: Bar...|
|Fri Dec 21 22:53:...| organicsblue| 1|Amazon Kindle Pap...|
|Fri Dec 21 22:53:...| Odielicious18| 729|Magna now, West J...|
|Fri Dec 21 22:53:...| ArticLightning| 168|I liked a @YouTub...|
|Fri Dec 21 22:53:...| skyriver1| 679|If social media w...|
|Fri Dec 21 22:53:...| MAliciavp| 435|RT @Clapsico: OTR...|
|Fri Dec 21 22:53:...| dave53elmer| 127|RT @meridak Obam...|
|Fri Dec 21 22:53:...| onegringaa| 331|RT @OceannJennaa:...|
|Fri Dec 21 22:53:...| FingerStinger| 5353|"Bred" definition...|
|Fri Dec 21 22:53:...| mario87s| 111|RT @atienzajc: Ll...|
|Fri Dec 21 22:53:...| brownandbella| 788|RT @Green_Footbal...|
|Fri Dec 21 22:53:...| zhaiickodrjv| 1604|He's playin' Call...|
+--------------------+---------------+-------------+--------------------+
only showing top 20 rows
// ¿Quien es el usuario con mayor número de seguidores?
// importar functions para poder ordenar
val tweetOrdenado = tweetsDS.orderBy(desc("numSeguidores"))
tweetOrdenado.select("usuario").limit(1).show()
+-----------+
| usuario|
+-----------+
|BarackObama|
+-----------+
tweetOrdenado: org.apache.spark.sql.Dataset[Tweet] = [fecha_creacion: string, usuario: string ... 2 more fields] - Btrue
- A// vamos a trabajar con datasets
59Haciendo 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.
- Aimport org.apache.spark.storage.StorageLevel
tweetsDS.persist(StorageLevel.DISK_ONLY)
import org.apache.spark.storage.StorageLevel
res32: tweetsDS.type = [fecha_creacion: string, usuario: string ... 2 more fields]
// ¿Cuántos tweets contienen la palabra "data"?.
val dataDS = tweetsDS.select("texto").map(t => t.toString().split(" "))
dataDS.show()
+--------------------+
| value|
+--------------------+
|[[Obama, meets, R...|
|[[@halluzaynating...|
|[[@hii1d, Awww., ...|
|[[All, you, need,...|
|[[RT, @PatDollard...|
|[[Crankin, to, th...|
|[[RT, @InjusticeF...|
|[[Prio, 2, TS500,...|
|[[RT, @BBCWorld:,...|
|[[Amazon, Kindle,...|
|[[Magna, now,, We...|
|[[I, liked, a, @Y...|
|[[If, social, med...|
|[[RT, @Clapsico:,...|
|[[RT, @meridak, ,...|
|[[RT, @OceannJenn...|
|[["Bred", definit...|
|[[RT, @atienzajc:...|
|[[RT, @Green_Foot...|
|[[He's, playin', ...|
+--------------------+
only showing top 20 rows
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") - Btrue
- Aimport org.apache.spark.storage.StorageLevel
60Crear 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.- Aval tweetsJson = spark.read.json("./documentos/large-17-json")
Intitializing Scala interpreter ...
Spark Web UI available at http://10.0.2.15:4041
SparkContext available as 'sc' (version = 2.3.0, master = local[*], app id = local-1532271294323)
SparkSession available as 'spark'
tweetsJson: org.apache.spark.sql.DataFrame = [contributors: string, coordinates: struct<coordinates: array<double>, type: string> ... 22 more fields]
// se genera la vista a partir del DataFrame
tweetsJson.createOrReplaceTempView("tweets")
val rt50 = spark.sql("SELECT first(retweeted_status.user.screen_name) AS screen_name, first(retweeted_status.text) AS text, sum(retweeted_status.retweet_count) AS retweet_count FROM tweets WHERE retweeted_status.retweet_count > 0 GROUP BY(text) ORDER BY retweet_count DESC limit 50")
rt50.show()
+--------------+--------------------+-------------+
| screen_name| text|retweet_count|
+--------------+--------------------+-------------+
| justinbieber|i kinda felt like...| 3978764|
| WeLoveFredo|This is me rn. ca...| 2633561|
| DaRealMayans|REBECCA BLACK TRI...| 1589543|
| justinbieber|haha. u guys are ...| 1097901|
|dailyteenwords|I want to live my...| 988142|
| soldier7078|Merry Xmas from t...| 797796|
| ohteenquotes|I'm not popular, ...| 784667|
| justinbieber|i love my beliebe...| 770604|
| justinbieber|http://t.co/tFBHqTp5| 715651|
| damnitstrue|I want to live my...| 679720|
|GhettoChiIdren|Black, White, Chi...| 629643|
| ForMyKidrauhl|"Harry, Who's the...| 626686|
| BarackObama|"There is absolut...| 626016|
| UberFacts|Cracking your knu...| 496624|
|TheOfficialTed|Poor bitch http:/...| 483533|
|GhettoChiIdren|Why Black People ...| 471489|
|HuffingtonPost|POLL: Retweet if ...| 444159|
| ForMyKidrauhl|Directioners, if ...| 437221|
| lmfaoharry|Fans: "OMG ONE DI...| 366203|
| AlfredoFlores|@LittlecBeadles T...| 340069|
+--------------+--------------------+-------------+
only showing top 20 rows
rt50: org.apache.spark.sql.DataFrame = [screen_name: string, text: string ... 1 more field] - Btrue
- Aval tweetsJson = spark.read.json("./documentos/large-17-json")
61Crear 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.
- A//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 - Btrue
- A//como se va a trabajar con ficheros se realizan las importaciones
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.- A//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 - Btrue
- A//como se va a trabajar con ficheros se realizan las importaciones
63Join
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.- Aimport 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 - Btrue
- Aimport org.apache.spark.sql.types._
64Ventanas Streaming
Crear RDD para ventana de datos de 10 segundos.- A//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) - Btrue
- A//se realizan las importaciones para manejo de DStream.
65DataFram y Dataset - Streaming
Ejemplo de cuenta palabras.- A//se realizan las importaciones para manejo de DStream.
import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.sql.SparkSession
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)
//se crea un cuenta palabras leyendo desde el socket
val palabras = lines.flatMap(_.split(" "))
//para cada RDD que forman el String realizar...
palabras.foreachRDD { rdd =>
//se crea variable SparkSession
val spark = SparkSession.builder.config(rdd.sparkContext.getConf).getOrCreate()
import spark.implicits._
//se crea DF con una columna "palabra"
val palabrasDataFrame = rdd.toDF("palabra")
//se crea una vista
palabrasDataFrame.createOrReplaceTempView("palabras")
//se realiza una query
val cuentas = spark.sql("SELECT palabra, count(*) AS total FROM palabras GROUP BY palabra")
cuenta.show()
}
//start para comenzar
ssc.start()
ssc.awaitTerminationOrTimeout(60000) - Btrue
- A//se realizan las importaciones para manejo de DStream.
66transformaciones más comunes:
- Amap
filter
flatMap
mapPartitions
mapPartitionsWithIndex
sample
union
intersection
distinct
groupByKey - Btrue
- Amap
67Las siguientes son las acciones más comunes:
- Areduce
collect
count
first
take
foreach
countByKey
takeSample
takeOrdened
saveAsTextFile
saveAsSequenceFile
saveAsObjectFile - Btrue
- Areduce
68RDD.map
- ADevuelve un nuevo RDD aplicando una función a cada elemento de este RDD.
val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9))
val rdd2 = rdd1.map(_ * 2)
rdd2.collect
res: Array[Int] = Array(2, 4, 6, 8, 10, 12, 14, 16, 18) - Btrue
- ADevuelve un nuevo RDD aplicando una función a cada elemento de este RDD.
69flatMap()
- Aval rdd1 = sc.parallelize(List("En un lugar de la mancha", "de cuyo nombre no quiero acordarme"))
val rdd2 = rdd1.map(x => x.split(" ")) // split(" ") returns an array of words
rdd2.collect()
val rdd3 = rdd1.flatMap(x => x.split(" "))
rdd3.collect()
res2: Array[Array[String]] = Array(Array(En, un, lugar, de, la, mancha), Array(de, cuyo, nombre, no, quiero, acordarme))
res3: Array[String] = Array(En, un, lugar, de, la, mancha, de, cuyo, nombre, no, quiero, - Btrue
- Aval rdd1 = sc.parallelize(List("En un lugar de la mancha", "de cuyo nombre no quiero acordarme"))
70RDD.filter
- ADevuelve un nuevo RDD que contiene solo los elementos que satisfacen un predicado.
scala> textFile.filter(line => line.contains("Spark")).count() // How many lines contain "Spark"?
res3: Long = 15 - Btrue
- ADevuelve un nuevo RDD que contiene solo los elementos que satisfacen un predicado.
71mapPartitions
- ADevuelve un nuevo RDD aplicando una función a cada partición de este RDD.
val rdd = sc.parallelize(List("a","b","c","d","e","f"),3)
def myfunc(index: Int, iter: Iterator[(String)]) : Iterator[String] = {
iter.map(x => "[id de partición:" + index + ", valor: " + x + "]")
}
rdd.mapPartitionsWithIndex(myfunc).collect
res: Array[String] = Array([id de partición:0, valor: a], [id de partición:0, valor: b], [i - Btrue
- ADevuelve un nuevo RDD aplicando una función a cada partición de este RDD.
72mapPartitionsWithIndex：
- AEsta función tiene el mismo efecto que mapPartitions, excepto que hay un parámetro de índice de partición más
var rdd1 = sc.makeRDD(1 to 5,2)//
// rdd1 tiene dos particiones
var rdd2 = rdd1.mapPartitionsWithIndex{
(x,iter) => {
var result = List[String]()
var i = 0
while(iter.hasNext){
i += iter.next()
}
result.::(x + "|" + i).iterator
}
}
// rdd2 acumula los números de cada partición en rdd1 y agrega el índice de partición delante del resultado de acumulación de cada partición
scala> rdd2.collect
res13: Array[String] = Array(0|3, 1|12) - Btrue
- AEsta función tiene el mismo efecto que mapPartitions, excepto que hay un parámetro de índice de partición más
73Sample
- ADevuelve un subconjunto muestreado de este RDD.
val a = sc.parallelize(1 to 10000, 3)
a.sample(false, 0.2).take(1000) - Btrue
- ADevuelve un subconjunto muestreado de este RDD.
74union()
- ANos devuelve la unión de los dos RDDs
val rdd1 = sc.parallelize(1 to 4, 1)
val rdd2 = sc.parallelize(8 to 10, 1)
rdd1.union(rdd2).collect()
res: Array[Int] = Array(1, 2, 3, 4, 8, 9, 10) - Btrue
- ANos devuelve la unión de los dos RDDs
75Unión por clave – join()
- ANos devuelve la unión de dos RDDs unidos por clave. Si los conjuntos de datos son de tipo (K, V) y (K,W), nos devuelve un conjunto de datos de (K, (V, W)) para cada clave.
val rdd1 = sc.parallelize(List((1, "tren"), (2, "avion"), (3, "barco"), (3, "moto")))
val rdd2 = sc.parallelize(List((1, "rojo"), (2, "azul"), (3, "verde")))
rdd1.join(rdd2).collect()
res: Array[(Int, (String, String))] = Array((1,(tren,rojo)), (2,(avion,azul)), (3,(barco,verde)), (3,(moto,verde))) - Btrue
- ANos devuelve la unión de dos RDDs unidos por clave. Si los conjuntos de datos son de tipo (K, V) y (K,W), nos devuelve un conjunto de datos de (K, (V, W)) para cada clave.
76Intersección – intersection()
- ANos devuelve la interseción de los dos RDDs
val rdd1 = sc.parallelize(1 to 6, 1)
val rdd2 = sc.parallelize(4 to 10, 1)
rdd1.intersection(rdd2).collect()
res: Array[Int] = Array(4, 6, 5) - Btrue
- ANos devuelve la interseción de los dos RDDs
77Elementos distintos de RDD – distinct()
- ANos devuelve un RDD con los elementos distintos
val rdd = sc.parallelize(List("avion", "avion", "tren", "tren", "tren", "barco"), 2)
rdd.distinct.collect()
res: Array[String] = Array(avion, tren, barco) - Btrue
- ANos devuelve un RDD con los elementos distintos
78Agrupar por clave – groupByKey()
- AAgrupa los elementos de un RDD por clave.
val words = sc.parallelize(List("avion", "tren", "barco", "coche", "moto", "bici"), 2)
val rdd_with_key = words.keyBy(_.length) // se usa la longitud de la palabra como clave
rdd_with_key.groupByKey.collect()
res: Array[(Int, Iterable[String])] = Array((4,CompactBuffer(tren, moto, bici)), (5,CompactBuffer(avion, barco, coche))) - Btrue
- AAgrupa los elementos de un RDD por clave.
79Reducir por clave – reduceByKey()
- AAgrega elementos de un RDD por clave de tal manera que estos queden aplanados
val rdd = sc.parallelize(List("avion", "tren", "barco", "coche", "moto", "bici"), 2)
val map_rdd = rdd.map(x => (x.length, x))
map_rdd.reduceByKey(_ + ";" + _).collect()
res: Array[(Int, String)] = Array((4,tren;moto;bici), (5,avion;barco;coche)) - Btrue
- AAgrega elementos de un RDD por clave de tal manera que estos queden aplanados
80Reduce () Accion
- AReduce los elementos de este RDD utilizando el operador binario conmutativo y asociativo especificado.
val listRdd = spark.sparkContext.parallelize(List(1,2,3,4,5,3,2))
println("output sum using binary : "+listRdd.reduce(_ min _))
println("output min using binary : "+listRdd.reduce(_ max _))
println("output max using binary : "+listRdd.reduce(_ + _))
output sum using binary : 1
output min using binary : 5
output max using binary : 20 - Btrue
- AReduce los elementos de este RDD utilizando el operador binario conmutativo y asociativo especificado.
81Collect accion
- ADevuelve una matriz que contiene todos los elementos de este RDD.
- Btrue
82Acciones
- Acount: nos devuelve el número total de elementos:
1
sc.parallelize(List(1, 2, 3, 4)).count //4
collect: nos vuelca toda la colección distribuida en un array en memoria:
1
sc.parallelize(List(1, 2, 3, 4)).collect // Array(1, 2, 3, 4)
Ojo, cuidao. Si el RDD es muy grande, podemos tener problemas al volcar toda la colección en memoria.
saveAsTextFile: nos vuelca la información en un fichero de texto:
1
intValues.saveAsTextFile("results.txt") - Btrue
- Acount: nos devuelve el número total de elementos:
Información
- Descripción
- Test sobre Sin título, creado por Carlos Alberto Ramirez Martinez el 23/10/2021.
- Idioma
- English
- Visitas
- 2
- Valoración
- Aún sin valoraciones Valora este test