Scala

Lo que a continuación vamos realizar es crear un script en scala y posteriomente ejecutarlo desde línea de comando como si estuvieramos realizando un proceso.
Abrimos nuestro IDE eclipse scala
spark
Seleccionamos nuestro workspace
spark
Seleccionamos File->New->Other...
Selecionamos Maven->Maven Project y presionamos Next
spark
Seleccionamos Create a simple project(skip archetype selection) y presionamos next spark
Ahora llenamos los siguientes datos:
y presionamos finish.
spark
Nos vamos al proyecto hacemos clic derecho y seccionamos Configure->Add Nature Scala y nos queda de la siguiente forma
spark
Hacemos clic derecho en el proyecto y Seleccionamos Build Path-<Configure Build Path...
spark
Seleccionamos la pestaña que dice source y presionamos Edit..
spark
El folder los renombramos src/main/scala y presionamos finish
spark
seleccionamos el directorio innertest/src/test/java y presionamos Remove
spark
seleccionamos el directorio innertest/src/test/resources y presionamos Remove
spark
Nos vamos a la pestaña que dice Libraries seleccionamos JRE System Library [J2SE-1.5] y presionamos Remove
spark
Posteriorme ahí presionamos Add Library en Add Library seleccionamos JRE System Library y presionamos Next
spark
Dejamos el jre que esta por defualt y presionamos finish
spark
Por ultimo Presionamos Apply y luego Apply and Close
spark
El archivo innertest/pom.xml queda de la siguiente forma:
       
         
 
         
         
   <project xmlns="http://maven.apache.org/POM/4.0.0"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.tutosoftware.innertest</groupId>
  <artifactId>innertest</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <name>Inner Join Spark Cassandra</name>
  <dependencies>
 <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.12</artifactId>
    <version>3.1.2</version>
</dependency>
  
 <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.12</artifactId>
    <version>3.1.2</version>
</dependency>
  
 
 <dependency>
    <groupId>com.datastax.spark</groupId>
    <artifactId>spark-cassandra-connector_2.12</artifactId>
    <version>3.1.0</version>
</dependency>
  
  
  
  </dependencies>
  
  <build>
        <sourceDirectory>src</sourceDirectory>
        <plugins>
            <plugin>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.5.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>

            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.3.1</version>
            </plugin>
        </plugins>
    </build>
  
  
  
</project>

en src/main/scalacreamos el paquete com.tutosoftare.innertest En ese paquete seleccionamos New->Scala Object
spark
Lo nombramos InnerMain y presionamos finish
spark
spark
Escribimos el siguiente código:
       
        
              
      

package com.tutosftware.innertest


import com.datastax.spark.connector._
import com.datastax.spark.connector.cql._
import org.apache.spark._
import org.apache.log4j._
import org.apache.spark.sql.SparkSession._
import org.apache.spark.sql.functions._

object InnerMain {
   def main(args:Array[String]) {
  
     val cassandraHost = "127.0.0.1"
     
     
     val conf = new SparkConf().setAppName("InnerJoinApp")
     .setMaster("local")
     .set("spark.cassandra.connection.host",cassandraHost)
     
     val sc = new SparkContext(conf)
     
     val sqlContext = new org.apache.spark.sql.SQLContext(sc)
     
     
      val proveedores = sqlContext.
    read.
    format("org.apache.spark.sql.cassandra").
    options(Map( "table" -> "proveedor", "keyspace" -> "productodb")).
    load()


proveedores.show()
     
     val productos = sqlContext.
    read.
    format("org.apache.spark.sql.cassandra").
    options(Map( "table" -> "producto", "keyspace" -> "productodb")).
    load()


productos.show()
     
 proveedores.createOrReplaceTempView("proveedor")
 productos.createOrReplaceTempView("producto")
 
 
 sqlContext.sql(" select a.idProducto,a.nombreProducto,b.nombreProveedor,b.telefono from producto a" 
     +"INNER JOIN proveedor b ON  a.idProveedor = b.idProveedor ").show()
     
     sc.stop()
     
     
     
       
     
     
     
   }
}

Lo que sigue es que vamos a generar nuestro jar con maven.Hacemos clic derecho en el proyecto selecionamos Run As->Maven Build spark
En Goals escribimos package presionamos Apply y luego Run
spark
En directorio target se genera el jar
spark
Abrimos una terminal y nos pasamos al directorio donde se genero nuestro jar y escribimos el siguiente comando:
       
       
       
       
       spark-submit --packages com.datastax.spark:spark-cassandra-connector_2.12:3.1.0 --class com.tutosftware.innertest.InnerMain 
 --master spark://127.0.0.1:7077 innertest-0.0.1-SNAPSHOT.jar

spark
spark
spark

Conclusión

Hay ocasiones que también requerimos de ejecutar procesos para sacar información de nuestros datos por eso realice el ejemplo operativamente y programacionalmente depende de tu necesidad a la hora de procesar datos ya que spark como motor de procesamiento combinado con cassandra pueden ser una herramienta valiosa para procesar un gran volumen de datos en tiempo real que es le puedes vender a tus clientes o las empresas.
Anterior