Hagamos un proyecto con scalatra como en los viejos tiempos

 Hace mucho mucho que no hago un proyecto con scalatra y me dio añoranza, así que vamos a hacer un "hola mundo" para rememorar buenos tiempos. 

Primero tenemos que tener instalado sbt, yo lo instale con sdkman (que si no lo conocen dejo el post acá) 

sdk i sbt 

y listo, luego creemos nuestro proyecto : 

sbt new scalatra/scalatra.g8

ahí nos va a pedir unos datos, complete nomas ...

Y bueno, ya esta... Porque scalatra nos trae un ejemplo el cual esta en el paquete com.example.app

ackage com.example.app

import org.scalatra._

class MyScalatraServlet extends ScalatraServlet {

  get("/") {
    views.html.hello()
  }

}

Que lo que hace es mostrar lo que esta en la vista es decir en src/main/twirl/views/hello.scala.html : 

@()
@layouts.html.default("Scalatra: a tiny, Sinatra-like web framework for Scala", "Welcome to Scalatra"){
  <p>Hello, Twirl!</p>
}

Que es una mezcla de scala con Html. 

Ahora vamos a ver el ejemplo, en marcha, en el directorio del proyecto, escribimos sbt y luego 

> jetty:start

Y la aplicación se verá en http://localhost:8080

Y ya esta!! 

Dejo link: https://scalatra.org

Spark y HBase

Tuve que conectar una aplicación spark con hbase con scala y sbt y scalatra... etc...

primero vamos a ver a build.sbt : 

val ScalatraVersion = "2.6.3"

organization := "com.hexacta"

name := "Sparklatra"

version := "0.1.0-SNAPSHOT"

scalaVersion := "2.11.8"

resolvers += Classpaths.typesafeReleases

resolvers += "Cloudera Repository" at "https://repository.cloudera.com/content/repositories/releases/"

resolvers += "Cloudera Repository2" at "https://repository.cloudera.com/artifactory/cloudera-repos/"

libraryDependencies ++= Seq(

  "org.scalatra" %% "scalatra" % ScalatraVersion  ,

  "org.scalatra" %% "scalatra-scalatest" % ScalatraVersion % "test" ,

  "ch.qos.logback" % "logback-classic" % "1.2.3" % "runtime" ,

  "org.eclipse.jetty" % "jetty-webapp" % "9.4.9.v20180320" % "container" ,

  "javax.servlet" % "javax.servlet-api" % "3.1.0" % "provided"  ,

  "com.sun.jersey" % "jersey-core" % "1.19.4" ,

  "com.sun.jersey" % "jersey-server" % "1.19.4" ,

  "org.apache.spark" % "spark-core_2.11" % "2.3.1",

  "org.apache.spark" %% "spark-sql" % "2.3.1",

  "org.apache.spark" %% "spark-streaming" % "2.3.1" ,

  "org.apache.hbase" % "hbase-spark" % "2.1.0-cdh6.1.1" intransitive() ,

  "org.apache.hbase" % "hbase-server" % "2.1.0-cdh6.1.1"  intransitive() ,

  "org.apache.hbase" % "hbase-mapreduce" % "2.1.0-cdh6.1.1" intransitive() ,

  "org.apache.hbase" % "hbase-common" % "2.1.0" exclude("com.fasterxml.jackson.core", "jackson-databind"),

  "org.apache.hbase" % "hbase-client" % "2.1.0" exclude("com.fasterxml.jackson.core", "jackson-databind")

)

enablePlugins(SbtTwirl)

enablePlugins(ScalatraPlugin)

Ahora vamos hacer un ejemplito, de un metodo get que insertar y consultar : 

   get(s"/testSparkHbase/:key/:value") {

    val conf : Configuration = HBaseContext.getConf()

    // cree la tabla

    // create 'TableTest', 'info'

    // put 'TableTest', 'rowkey1', 'info:test', 'ejemplo'

    val connection = ConnectionFactory.createConnection(conf)

    val hbaseContext = new org.apache.hadoop.hbase.spark.HBaseContext(SparkContext.getSc, conf)

    val tableName = "TableTest"

    val columnFamily = "info"

    val rdd = SparkContext.getSc.parallelize(Array(

      (Bytes.toBytes(params("key")),

        Array((Bytes.toBytes(columnFamily), Bytes.toBytes("test"), Bytes.toBytes(params("value")))))))

    hbaseContext.bulkPut[(Array[Byte], Array[(Array[Byte], Array[Byte], Array[Byte])])](rdd,

      TableName.valueOf(tableName),

      (putRecord) => {

        val put = new Put(putRecord._1)

        putRecord._2.foreach((putValue) =>

          put.addColumn(putValue._1, putValue._2, putValue._3))

        put

      });

    Ok("ok")

  }

  get(s"/testSparkHbase/:key") {

    val conf : Configuration = HBaseContext.getConf()

    // cree la tabla

    // create 'TableTest', 'info'

    // put 'TableTest', 'rowkey1', 'info:test', 'ejemplo'

    val connection = ConnectionFactory.createConnection(conf)

    val hbaseContext = new org.apache.hadoop.hbase.spark.HBaseContext(SparkContext.getSc, conf)

    val tableName = "TableTest"

    val columnFamily = "info"

    val rdd = SparkContext.getSc.parallelize(Array(Bytes.toBytes(params("key"))))

    val getRdd = rdd.hbaseBulkGet[String](hbaseContext, TableName.valueOf(tableName), 2,

      record => {

        System.out.println("making Get"+ record.toString)

        new Get(record)

      },

      (result: Result) => {

        val it = result.listCells().iterator()

        val b = new StringBuilder

        b.append(Bytes.toString(result.getRow) + ":")

        while (it.hasNext) {

          val cell = it.next()

          val q = Bytes.toString(CellUtil.cloneQualifier(cell))

          if (q.equals("counter")) {

            b.append("(" + q + "," + Bytes.toLong(CellUtil.cloneValue(cell)) + ")")

          } else {

            b.append("(" + q + "," + Bytes.toString(CellUtil.cloneValue(cell)) + ")")

          }

        }

        b.toString()

      })

    getRdd.collect().foreach(v => println(v))

    var result = ListBuffer[String]()

    getRdd.collect().foreach(v => result += v)

    Ok(compact(render(result)))

  }

  }

Dejo el git: https://github.com/emanuelpeg/spark-scalatra

Insertar datos en Apache HBase con Scala y Scalatra

Vamos hacer un ejemplo de una API Rest con scalatra que inserte datos en HBase.

Primero hacemos un proyecto en scalatra como ya hemos explicado:
https://emanuelpeg.blogspot.com/2018/08/haciendo-un-proyecto-con-scalatra.html

Luego agregamos las dependencias de hbase, en el archivo build.sbt agregando :

libraryDependencies ++= Seq(
  "org.apache.hbase" % "hbase-common" % "2.1.0" ,
  "org.apache.hbase" % "hbase-client" % "2.1.0"
)

Luego en la clase de ejemplo generamos el método get que inserta datos en la base hbase :

 get(s"/guardar/:rowKey/:str") {

    val conf : Configuration = HBaseConfiguration.create()
    // Se debe agregar en el archivo host el nombre de la base
    conf.set("hbase.zookeeper.quorum", "mybase")
    conf.set("hbase.zookeeper.property.clientPort", "2181")

    // seteamos los puertos de hbase.
    conf.set("hbase.master.port", "60000")
    conf.set("hbase.master.info.port", "60010")
    conf.set("hbase.regionserver.info.port", "60030")
    conf.set("hbase.regionserver.port", "60020")
    // cree la tabla
    // create 'TableTest', 'info'
    // put 'TableTest', 'rowkey1', 'info:test', 'ejemplo'
    val connection = ConnectionFactory.createConnection(conf)
    val table = connection.getTable(TableName.valueOf( "TableTest" ) )

    // Put example
    val put = new Put(Bytes.toBytes(params("rowKey")))
    put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("test"), Bytes.toBytes(params("str")))

    table.put(put)

    Ok("ok")
  }

}

El método inserta los datos pasados por parámetro en la URL. Donde pasamos una key y un dato. De esta manera insertamos datos en una tabla TableTest.

Algo importante que se puede ver es que seteo los puestos dado que en las ultimas versiones cambiaron de  600X0 a 160X0. Depende de la versión de Hbase, hay que usar uno u otro, yo estoy usando HBase de la maquina virtual de cloudera.

Y Listo!!

Haciendo un proyecto con scalatra y spark

Vamos a hacer un proyecto en Scala con Scalatra y Spark. La idea es hacer una Api Rest, la cual utilice Spark.

Lo primero que hacemos es el proyecto de scalatra como esta indicado aquí :
https://emanuelpeg.blogspot.com/2018/08/haciendo-un-proyecto-con-scalatra.html

Ya teniendo este proyecto agregamos las dependencias de spark y de jersey (este ultimo es necesario para el funcionamiento de los frameworks)

Es decir el buil.sbt nos va quedar así:

val ScalatraVersion = "2.6.3"

organization := "com.hexacta"

name := "Sparklatra"

version := "0.1.0-SNAPSHOT"

scalaVersion := "2.11.0"

resolvers += Classpaths.typesafeReleases

libraryDependencies ++= Seq(
  "org.scalatra" %% "scalatra" % ScalatraVersion  ,
  "org.scalatra" %% "scalatra-scalatest" % ScalatraVersion % "test" ,
  "ch.qos.logback" % "logback-classic" % "1.2.3" % "runtime" ,
  "org.eclipse.jetty" % "jetty-webapp" % "9.4.9.v20180320" % "container" ,
  "javax.servlet" % "javax.servlet-api" % "3.1.0" % "provided"  ,
  "com.sun.jersey" % "jersey-core" % "1.19.4" ,
  "com.sun.jersey" % "jersey-server" % "1.19.4" ,
  "org.apache.spark" %% "spark-core" % "2.3.1"
)

enablePlugins(SbtTwirl)
enablePlugins(ScalatraPlugin)

Luego de agregar estas dependencias vamos a hacer un objeto que contenga el contexto de Spark, dado que queremos utilizar un contexto en toda nuestra aplicación y lo hacemos de la siguiente manera :

package com.miEmpresa

import org.apache.spark.{SparkConf, SparkContext}

object SparkContext {

  //Create a SparkContext to initialize Spark
  val conf = new SparkConf()
  conf.setMaster("local")
  conf.setAppName("Word Count")
  val sc = new SparkContext(conf)

   def getSc = {
      sc
   }
}

Luego de hacer esto, vamos a servlet de scalatra y agregamos el método get que cuente palabras pasadas como parámetro en la url :

   get(s"/contar/:str") {

    //word count
    val counts = List({
      params("str")
    }).flatMap(line => line.split(" "))
      .map(word => (word, 1))

    val countsRdd = SparkContext.getSc.parallelize(counts).reduceByKey(_ + _).collect()

    var result = ListBuffer[(String,Int)]()

    countsRdd.foreach(line => result += line)

    Ok(compact(render(result)))

  }

En el get definimos que la url va a contener un string que es nuestro parámetro str. Luego hacemos un split por espacios en blanco y luego hacemos un mapa (palabra, 1).  Como ultimo paso convertimos nuestro map a RDD y reducimos por palabra. Es decir el proceso sería para la linea "hola hola mundo" de la siguiente manera :

 "hola hola mundo" -> split(" ") = ("hola","hola","mundo") -> map = ( ("hola", 1) , ("hola", 1) , ("mundo", 1) ) -> reduceByKey = ("hola",2), ("mundo",1)

Para probarlo primero debemos ejecutarlo con sbt y jetty. Vamos al directorio donde se encuentra el archivo build.sbt y escribimos:
sbt
--- Acá va correr un montón de cosas ---
jetty:start

Por ultimo, debemos ir a un browser y probar la url : http://localhost:8080/contar/

Como se envían los datos por el método get podemos probar con la siguiente url, por ejemplo:

"http://localhost:8080/contar/hola hola hola Mundo"

Y listo!!

Haciendo un proyecto con scalatra

Scalatra es el Sinatra de scala.

Para el que no conoce Sinatra es un framework que permite hacer paginas web con las menores configuraciones y desarrollo posible. Trata de acelerar el desarrollo manteniendo un modelo simple y convención antes de configuración.  Y el mismo espíritu tiene Scalatra.

Vamos hacer un proyecto con sbt.

Vamos a la consola y hacemos:

sbt new scalatra/scalatra.g8

Luego nos va pedir un montón de datos, pero es fácil saber a que se refiere dado que presenta unos ejemplos.

Y con eso ya estamos, podemos ejecutar esto con el comando :

sbt jetty:start

Y podemos chequearlo en la url: http://localhost:8080/

Como funciona?

La clase MyScalatraServlet capta el método get / y responde con la vista.

import org.scalatra._

class MyScalatraServlet extends ScalatraServlet {

  get("/") {
    views.html.hello()
  }

}

Dejo link: http://scalatra.org/