Observatorio BI & Analytics

La deshumanización de la IA

noreply@blogger.com (fglopez) — Fri, 04 Jul 2025 22:00:00 +0000

Vivimos tiempos de un fervor desmedido por la inteligencia artificial. Sus promesas se gritan en todos los foros: automatización, eficiencia, velocidad. Pero, ¿a qué precio? Tras esa fachada de progreso y soluciones instantáneas se oculta un proceso silencioso de deshumanización que afecta, ante todo, a quienes sostienen en sus hombros la solidez de los proyectos: los ingenieros senior.

Las consultoras han encontrado en la IA un arma de doble filo. La utilizan para justificar la imposición de tiempos de entrega imposibles, argumentando que “la IA te lo hace todo”. Pero la realidad es muy distinta. La IA comete errores, su salida es aproximada, a veces incluso absurda o dañina. Es necesaria la revisión, la reflexión y el ajuste experto para transformar sus propuestas en soluciones reales. Sin embargo, las empresas ya no quieren oír hablar de reflexión ni de maduración conceptual. Quieren resultados en tres semanas. Sprint tras sprint. Ciclos sin pausa que convierten al ingeniero en un ratón atrapado en una rueda que no cesa de girar, mientras su cuerpo y su mente piden descanso y su dignidad clama por ser reconocida.

La IA, tal como se implementa hoy, no es una herramienta formativa para las nuevas generaciones. Los ingenieros junior, al estar aún en su etapa de aprendizaje, no siempre tienen la experiencia suficiente para evaluar la validez de lo que la IA propone. Requieren un proceso humano de enseñanza, razonamiento y error guiado que la IA no proporciona. Así, la carga se traslada inexorablemente a los seniors, que deben ejercer de revisores, arquitectos y escudos ante los fallos. Su jornada se convierte en una sucesión de soluciones superficiales que deben recomponer en silencio para salvar la entrega.

Mientras tanto, la creatividad se asfixia. Ya no hay espacio para pensar de forma original, para innovar o para debatir. La IA ofrece respuestas inmediatas que se convierten en la verdad sin discusión, aunque sean erróneas, aunque carezcan de contexto. El ingeniero pasa de ser un creador a un corrector, de un pensador a un ejecutor, de un arquitecto de soluciones a un editor de ocurrencias generadas por un algoritmo estadístico entrenado en datos mediocres.

Y así, el círculo vicioso se cierra: los seniors, agotados y desencantados, se irán. Se jubilarán. Buscarán otros caminos. Y los juniors, sin haber recibido la formación y la práctica necesarias para desarrollar su criterio y su pensamiento crítico, no podrán suplirlos. Nos quedaremos con respuestas automáticas y sin mentes capaces de cuestionarlas. La IA solo es útil en manos de expertos, pero si no cultivamos nuevos expertos, la utilidad misma de la IA quedará sepultada en un desierto de respuestas no comprendidas.

Estamos ante la era de la inmediatez, donde el progreso tecnológico corre más rápido que la ética, donde se exalta la máquina y se olvida al humano, donde se produce sin pensar y se decide sin razonar.

La deshumanización de la IA no radica en que sea fría o mecánica. Radica en que nos obliga a serlo a nosotros.

Guía Desarrollo Spark

noreply@blogger.com (fglopez) — Fri, 15 Nov 2024 10:36:00 +0000

Funciones de Array de Spark SQL

https://sparkbyexamples.com/spark/spark-sql-array-functions/

Funciones de Map de Spark SQL

https://sparkbyexamples.com/spark/spark-sql-map-functions/

Funciones de primer orden

https://docs.databricks.com/en/semi-structured/higher-order-functions.html

Funciones integradas en Spark SQL

https://sparkbyexamples.com/spark/spark-sql-functions/#google_vignette

Tipos de Join en Spark SQL

https://sparkbyexamples.com/spark/spark-sql-dataframe-join/#sql-full-outer-join

Funciones de Ventana

https://sparkbyexamples.com/spark/spark-sql-window-functions/

Pivotado de un DataFrame Spark

https://sparkbyexamples.com/spark/how-to-pivot-table-and-unpivot-a-spark-dataframe/

Spark Performance Tuning

https://sparkbyexamples.com/spark/spark-performance-tuning/

Spark SQL Performance Tuning by Configurations

https://sparkbyexamples.com/spark/spark-sql-performance-tuning-configurations/

Tabla de Monitorización de tareas semanales

noreply@blogger.com (fglopez) — Mon, 04 Sep 2023 11:11:00 +0000

Septiembre 2023

Semana	Miembro Asignado	Tarea Completada (✓/✗)
1 - 7 Sept	Ana	✓
8 - 14 Sept	Carlos	✓
15 - 21 Sept	Beatriz
22 - 28 Sept	Diego

Octubre 2023

Semana	Miembro Asignado	Tarea Completada (✓/✗)
1 - 7 Oct	Elena
8 - 14 Oct	Francisco
15 - 21 Oct	Ana
22 - 28 Oct	Carlos

Noviembre 2023

Semana	Miembro Asignado	Tarea Completada (✓/✗)
1 - 7 Nov	Beatriz
8 - 14 Nov	Diego
15 - 21 Nov	Elena
22 - 28 Nov	Francisco

Nota: Una vez completada la tarea, por favor marque con un "✓" en la columna "Tarea Completada".
Recordatorio: Si no puedes realizar tu turno, coordina el cambio con un compañero y actualiza el tablón.

Plantilla de Peticiones JIRA para Data Engineering

noreply@blogger.com (fglopez) — Mon, 28 Aug 2023 08:38:00 +0000

1. Información Básica:

Título del JIRA: [Breve descripción del requerimiento o tarea]
Tipo de Tarea: [Análisis, Ingesta, Transformaciones Launcher, Transformaciones Capa Snapshot, Flujo CEFEO, Spark-Lineage/Datahub, Dolphin, Documentación, Otras Tareas Técnicas]

2. Descripción Detallada:

Objetivo: ¿Cuál es el propósito principal de esta tarea? ¿Qué se espera lograr?
Contexto: ¿Por qué es necesaria esta tarea? ¿Hay algún antecedente o situación actual que deba ser conocido?
Especificaciones Técnicas: Detalles técnicos relevantes para la tarea. Por ejemplo, para una ingesta, especificar la fuente de datos, el formato, la frecuencia, etc.

3. Recursos y Referencias:

Artefactos de Diseño: Enlaza directamente a los diseños técnicos, plantillas de ingestas, etc., preferiblemente almacenados en Confluence o en un repositorio centralizado.
Referencias Anteriores: Enlaza a JIRAs relacionados, correos electrónicos, conversaciones de Slack, etc., que puedan proporcionar contexto adicional.

4. Herramientas y Servicios AWS Involucrados:

Servicios AWS: [S3, Athena, Glue, EMR, etc.]
Lenguaje de Programación: [Scala, si es relevante para la tarea]

5. Criterios de Aceptación (Definition of Done):

Enumera los criterios específicos que deben cumplirse para considerar que la tarea está completa. Por ejemplo:
- Código desarrollado y revisado.
- Pruebas unitarias y de integración realizadas.
- Documentación actualizada.
- Datos cargados correctamente en S3 (para tareas de ingesta).
- Transformaciones validadas y datos disponibles en Athena (para tareas de transformación).
- Etc.

6. Comentarios Adicionales:

Cualquier otra información o detalle que el equipo deba conocer.

Ejemplo:

Campo	Detalle
Información Básica
Título del JIRA	Ingesta de datos de ventas desde API externa
Tipo de Tarea	Ingesta
Descripción Detallada
Objetivo	Ingestar datos de ventas diarias desde una API externa para análisis en Athena.
Contexto	Necesitamos analizar las ventas diarias en comparación con las campañas de marketing.
Especificaciones Técnicas	Fuente: API externa (URL proporcionada). Formato: JSON. Frecuencia: Diaria a las 00:00 UTC.
Recursos y Referencias
Artefactos de Diseño	[Enlace a Confluence con diseño técnico]
Referencias Anteriores	[Enlace a JIRA relacionado]
Herramientas y Servicios AWS
Servicios AWS	S3, Athena, Glue
Lenguaje de Programación	Scala
Subtareas y Estado
Subtarea 1	Desarrollo del código de ingesta 🟢
Subtarea 2	Pruebas unitarias 🟡
Subtarea 3	Carga de datos en S3 🟢
Subtarea 4	Validación en Athena 🟠
Subtarea 5	Actualización de documentación 🟡
Criterios de Aceptación (Definition of Done)
Criterio 1	Código desarrollado y revisado.
Criterio 2	Datos cargados correctamente en S3.
Criterio 3	Transformaciones validadas y datos disponibles en Athena.
Criterio 4	Documentación actualizada en Confluence.
Comentarios Adicionales	La API externa tiene un límite de 10,000 registros por solicitud.

Leyenda de Banderas de Colores:

🟢 Verde: Completado
🟡 Amarillo: En progreso
🟠 Naranja: Pendiente
🔴 Rojo: Bloqueado

Impresión de esquemas complejos de un Dataframe Spark

noreply@blogger.com (fglopez) — Mon, 14 Aug 2023 16:13:00 +0000

La función printSchema para que sea más general. Ahora acepta cualquier DataType y un prefijo. Si se encuentra con un StructType, recorre sus campos y llama a printSchema de nuevo con un nuevo prefijo. Si se encuentra con un ArrayType, llama a printSchema con el tipo de elemento del array y un nuevo prefijo con []. Si no es ninguno de esos dos casos, simplemente imprime el prefijo y el tipo.

Con este enfoque, el programa puede manejar estructuras anidadas complejas, ya sean campos STRUCT anidados dentro de otros campos STRUCT o ARRAY, o campos ARRAY anidados dentro de campos STRUCT.

Para generar un archivo CSV se puede usar la API de Spark para escribir DataFrames en formato CSV. Primero, la función printSchema devuelve una lista de tuplas (campo, tipo), y luego convertimos esa lista en un DataFrame que finalmente escribimos en formato CSV.

import org.apache.spark.sql.{SparkSession, DataFrame, Row} import org.apache.spark.sql.types._ object SchemaExplorer { def main(args: Array[String]): Unit = { val spark = SparkSession.builder .appName("SchemaExplorer") .master("local[*]") .getOrCreate() val df = spark.read.json("path_a_tu_esquema_json.json") val schemaList = extractSchema(df.schema) val outputDf = spark.createDataFrame(schemaList.map(Row.fromTuple), StructType(Seq( StructField("Campo", StringType, false), StructField("Tipo", StringType, false) ))) outputDf.write.csv("path_donde_guardar.csv") spark.stop() } def extractSchema(dataType: DataType, prefix: String = ""): List[(String, String)] = { dataType match { case structType: StructType => structType.fields.flatMap { field => val newPrefix = if (prefix.isEmpty) field.name else s"$prefix.${field.name}" extractSchema(field.dataType, newPrefix) }.toList case arrayType: ArrayType => val innerType = arrayType.elementType val newPrefix = s"$prefix[]" extractSchema(innerType, newPrefix) case _ => List((prefix, dataType.simpleString)) } } }

Vamos a suponer que tienes un DataFrame con un esquema que tiene un campo users que es un ARRAY de STRUCT. Cada estructura tiene campos name y address, donde address es otro STRUCT con campos street y zip.

val data = Seq( Row(Seq(Row("John", Row("123 Main St", "12345")), Row("Jane", Row("456 Elm St", "67890")))) ) val addressSchema = StructType(Seq( StructField("street", StringType, false), StructField("zip", StringType, false) )) val userSchema = StructType(Seq( StructField("name", StringType, false), StructField("address", addressSchema, false) )) val schema = StructType(Seq( StructField("users", ArrayType(userSchema), false) )) val df = spark.createDataFrame(spark.sparkContext.parallelize(data), schema)

Si ejecutamos el programa SchemaExplorer con este DataFrame, la salida sería:

Campo Tipo ------------------------- users[] StructType users[].name StringType users[].address StructType users[].address.street StringType users[].address.zip StringType

El programa se da cuenta de que users es un ARRAY y, por lo tanto, añade [] a su nombre. A continuación, como el tipo de elemento del ARRAY es un STRUCT, profundiza en sus campos y sigue haciendo lo mismo hasta que haya recorrido todo el esquema.

POO en Scala: traits, mixins y herencia

noreply@blogger.com (fglopez) — Sat, 29 Jul 2023 18:34:00 +0000

Object-oriented programming (OOP) is a popular programming paradigm that focuses on creating objects that can interact with each other to achieve specific tasks. Scala is a modern programming language that supports both object-oriented and functional programming paradigms. In this blog, we’ll explore some of the key object-oriented concepts in Scala, including traits, mixins, and inheritance.

CLASSES AND OBJECTS

Like other OOP languages, Scala has classes and objects. Classes define the blueprint for creating objects, which are instances of a class. Here’s an example of a simple class in Scala:

class Person(var name: String, var age: Int) {
  def sayHello(): Unit = {
    println(s"Hello, my name is $name and I'm $age years old.")
  }
}

This class defines a Person with a name and an age, and a sayHello() method that prints a greeting message.

To create an instance of the Person class, we can use the new keyword, like this:

val person = new Person("Alice", 30)

TRAITS

A trait is similar to an interface in Java. It defines a set of methods and fields that a class can implement. However, unlike Java interfaces, a trait can also provide a default implementation for these methods. Here’s an example:

trait Speaker {
  def speak(message: String): Unit = {
    println(message)
  }
}

class Dog extends Speaker {
  def bark(): Unit = {
    speak("Woof!")
  }
}

val dog = new Dog()
dog.bark() // prints "Woof!"

In this example, we define a trait called Speaker that has a single method called speak. We then define a class called Dog that extends the Speaker trait and implements the bark method, which calls the speak method to output a message.

Traits can be mixed in with classes using the with keyword. For example:

trait Runner {
  def run(distance: Double): Unit
}

class Athlete(val name: String) extends Runner with Speaker {
  def run(distance: Double): Unit = {
    speak(s"$name is running $distance meters")
  }
}

val athlete = new Athlete("John")
athlete.run(100.0) // prints "John is running 100.0 meters"

In this example, we define a trait called Runner that has a single method called run. We then define a class called Athlete that extends both the Runner and Speaker traits. This allows us to mix in multiple traits into a single class.

MIXINS

A mixin is a way of combining multiple traits together to form a new class. Mixins can be used to add functionality to a class without having to create a new inheritance hierarchy. Here’s an example:

trait Swimmer {
  def swim(distance: Double): Unit
}

trait Cyclist {
  def cycle(distance: Double): Unit
}

class Triathlete(val name: String) extends Swimmer with Cyclist with Speaker {
  def swim(distance: Double): Unit = {
    speak(s"$name is swimming $distance meters")
  }

  def cycle(distance: Double): Unit = {
    speak(s"$name is cycling $distance kilometers")
  }
}

val triathlete = new Triathlete("Jane")
triathlete.swim(100.0) // prints "Jane is swimming 100.0 meters"
triathlete.cycle(10.0) // prints "Jane is cycling 10.0 kilometers"

In this example, we define two traits called Swimmer and Cyclist. We then define a class called Triathlete that mixes in these traits along with the Speaker trait. This allows us to create a new class with the combined functionality of all three traits.

Inheritance

Inheritance is a way of creating a new class that is a modified version of an existing class. In Scala, a class can only inherit from a single class, but it can mix in multiple traits. Here’s an example:

class Animal(val name: String) {
  def speak(): Unit = {
    println("...")
  }
}

class Cat(name: String) extends Animal(name) {
  override def speak(): Unit = {

CONCLUSION

Scala’s support for OOP features like classes, objects, inheritance, traits, and mixins makes it a powerful language for building complex, object-oriented applications. By understanding these features, you can write more modular, reusable, and extensible code.

Crear bucle for/yield en Scala

noreply@blogger.com (fglopez) — Sat, 29 Jul 2023 18:22:00 +0000

This is an excerpt from the Scala Cookbook (#ad) (partially modified for the internet). This is Recipe 3.4, “How to create a ‘for comprehension’ (for/yield loop).”

Problem

You want to create a new Scala collection from an existing collection by applying an algorithm (and potentially one or more guards) to each element in the original collection.

Solution

Use a yield statement with a for loop and your algorithm to create a new collection from an existing collection.

For instance, given an array of lowercase strings:

scala> val names = Array("chris", "ed", "maurice")
names: Array[String] = Array(chris, ed, maurice)

you can create a new array of capitalized strings by combining yield with a for loop and a simple algorithm:

scala> val capNames = for (e <- names) yield e.capitalize
capNames: Array[String] = Array(Chris, Ed, Maurice)

Using a for loop with a yield statement is known as a for-comprehension. If your algorithm requires multiple lines of code, perform the work in a block after the yield keyword:

scala> val lengths = for (e <- names) yield {
     |     // imagine that this required multiple lines of code
     |     e.length
     | }
lengths: Array[Int] = Array(5, 2, 7)

Except for rare occasions, the collection type returned by a for comprehension is the same type that you begin with. For instance, if the collection you’re looping over is an ArrayBuffer:

var fruits = scala.collection.mutable.ArrayBuffer[String]()
fruits += "apple"
fruits += "banana"
fruits += "orange"

the collection your loop returns will also be an ArrayBuffer:

scala> val out = for (e <- fruits) yield e.toUpperCase
out: scala.collection.mutable.ArrayBuffer[java.lang.String] = ArrayBuffer(APPLE, BANANA, ORANGE)

If your input collection is a List, the for/yield loop will return a List:

scala> val fruits = "apple" :: "banana" :: "orange" :: Nil
fruits: List[java.lang.String] = List(apple, banana, orange)

scala> val out = for (e <- fruits) yield e.toUpperCase
out: List[java.lang.String] = List(APPLE, BANANA, ORANGE)

Discussion

If you’re new to using yield with a for loop, it can help to think of the loop like this:

When it begins running, the for/yield loop immediately creates a new, empty collection that is of the same type as the input collection. For example, if the input type is a Vector, the output type will also be a Vector. You can think of this new collection as being like a bucket.
On each iteration of the for loop, a new output element is created from the current element of the input collection. When the output element is created, it’s placed in the bucket.
When the loop finishes running, the entire contents of the bucket are returned.

That’s a simplification of the process, but I find it helpful when explaining the process. Writing a basic for/yield expression without a guard is just like calling the map method on a collection. For instance, the following for comprehension converts all the strings in the fruits collection to uppercase:

scala> val out = for (e <- fruits) yield e.toUpperCase
out: List[String] = List(APPLE, BANANA, ORANGE)

Calling the map method on the collection does the same thing:

scala> val out = fruits.map(_.toUpperCase)
out: List[String] = List(APPLE, BANANA, ORANGE)

When I first started learning Scala, I wrote all of my code using for/yield expressions until the map light bulb went on one day.

Pattern matching en Scala

noreply@blogger.com (fglopez) — Sat, 29 Jul 2023 18:04:00 +0000

Pattern matching, es un mecanismo usado en programación funcional que permite evaluar un objeto/valor/expresión contra una serie de valores/objetos/expresiones, es como un switch con esteroides ya que un switch sólo soporta algunos tipos en la evaluación (cadenas y enteros) por lo que resulta ser una versión más poderosa que el switch de Java o incluso que usar una serie de sentencias if-else, porque la coincidencia exitosa del valor contra el patrón definido puede deconstruir el valor en sus partes constituyentes.

Scala match vs Java switch

El pattern matching permite deconstruir un objeto y evaluarlo.

Veamos un ejemplo:

import scala.util.Randomval x: Int = Random.nextInt(10)x match {
  case 0 => "zero"
  case 1 => "one"
  case 2 => "two"
  case _ => "many"
}

En éste ejemplo podemos observar el uso del pattern matching en un valorpara deconstruir ese valor y “reaccionar” a él. La sintaxis es selector match { case clauses } donde selector es el valor y las case clauses son las alternativas que queremos comparar para hacer match. Es posible definir una expresión match y definir un valor de retorno de la siguiente forma:

def matchTest(x: Int): String = x match {   
  case 1 => "one"   
  case 2 => "two"   
  case _ => "many" 
}matchTest(3)  // many 
matchTest(1)  // one

De esta forma podemos reutilizar la expresión y usar su valor de retorno en otra expresión. Pattern matching se puede utilizar en distintos tipos de patrones (case clauses) como: constantes, variables, secuencias, tuplas, tipos, case classes, etc. Veamos algunos ejemplos de esto:

Sequences (Lists o Arrays)

expr match { 
  case List(0, _, _) => println("found it")
  case _ =>
}expr match { // use _* to avoid specifying how long it can be
    case List(0, _*) => println("found it")
    case _ =>
}

Tuples

def tupleDemo(expr: Any) = expr match {
  case (a, b, c)  =>  println("matched "+ a + b + c)
  case _ => 
}tupleDemo(("a ", 3, "-tuple")) // prints matched a 3-tuple

Types

def generalSize(x: Any) = x match {
    case s: String => s.length
    case m: Map[_, _] => m.size
    case _ => -1
}generalSize("abc) // returns Int = 3
generalSize(Map(1 -> 'a', 2 -> 'b')) // returns Int = 2

En los ejemplos anteriores se describe la simplicidad con la que podemos hacer uso de esta herramienta para diferentes tipos, la variedad de patrones que podemos encontrar puede enriquecer y simplificar lo que deseamos expresar en nuestro código, como el caso de los “types”, donde hacemos match por el tipo de dato para crear una expresión generalizada de tamaños de datos.

Ahora veamos un ejemplo con case clases.

abstract class Notificationcase class Email(sender: String, title: String, body: String) extends Notificationcase class SMS(caller: String, message: String) extends Notificationcase class VoiceRecording(contactName: String, link: String) extends Notificationdef showNotification(notification: Notification): String = {
  notification match {
  case Email(email, title, _) =>
      s"You got an email from $email with title: $title"
  case SMS(number, message) =>
      s"You got an SMS from $number! Message: $message"
  case VoiceRecording(name, link) =>
      s"you received a Voice Recording from $name! Click the link to hear it: $link"
  case _ => "No notification" 
  }
}
val someSms = SMS("12345", "Are you there?")
val someVoiceRecording = VoiceRecording("Tom", "voicerecording.org/id/123")println(showNotification(someSms))  // prints You got an SMS from 12345! Message: Are you there?println(showNotification(someVoiceRecording))  // you received a Voice Recording from Tom! Click the link to hear it: voicerecording.org/id/123

Pattern matching en case classes es muy útil y dota de mucha expresión sobre lo que queremos hacer con poco código, además de que no sólo podemos hacer case clauses sobre el tipo de valor sino sobre su estructura como es el caso de case Email(...) donde no nos importa el body del email sino sólamente hacer match con el email y el título, de esta forma podemos elegir sobre qué atributos hacer match. Otra cosa a destacar es la forma tan simple en la que hemos creado un método o función para visualizar notificaciones, un problema muy común hoy en día, pues ¿en cuántas páginas en internet o aplicaciones en nuestro celular no hemos visto notificaciones aparecer?, todas de distintos tipos pero siempre visualizadas de la misma forma, sólo basta pensar en un modelo de datos tan simple como el que muestra el ejemplo y lo demás viene por añadidura en el lenguaje.

Y así como creamos case clauses para case classes, podemos hacerlo para otros tipos siendo pattern matching una herramienta tan conveniente pues todoslos tipos pueden figurar para hacer match. Un ejemplo de esto puede ser el uso de pattern matching para manejar nuestras excepciones de una forma más funcional, en vez de sólo lanzar la excepción y terminar de forma inadecuada podemos crear case clauses que retornen un valor equivalente al error que estamos encontrando. Podemos “validar” el contenido de una variable, para saber si una función está recibiendo los parámetros adecuados o también el valor de retorno de una función por ejemplo en el caso de ser de tipo Option donde simplemente indicamos que puede contener “algo” lo cual sería un tipo Some, o no contener “algo” que sería un None, pero esto podemos verlo con más detalle en un siguiente post sobre Scala.

Por último quiero explicar los case clauses de los ejemplos donde agregamos un _ (guión bajo), esta notación llamada wildcard hace match con cualquier objeto en absoluto y puede servir de distintas formas según sea la necesidad, la principal puede ser simplemente que queremos descartar cualquier otro valor, todo lo que no haga match con los casos anteriores hará match con el wildcard, en nuestro último ejemplo se utiliza para determinar si el valor enviado a la función showNotification() no es una notificación y al tener como valor de retorno un String simplemente retornamos un mensaje que dice que no es una notificación.

¿Te diste cuenta de la notación en los Strings que creamos para cada caso? s”Message $value some other $value2”, en Scala esta notación es conocida como string interpolation que simplemente nos permite embeber la referencia de una variable o valor dentro de las literales String (una forma elegante de crear Strings), y esto es posible gracias a el interpolator “s” que está antes de la literal; existen otros interpolators y nosotros podríamos crear el nuestro :)

Pattern matching, Options, wildcards, interpolators, son sólo algunas cosas más que hacen a Scala un lenguaje elegante, y más allá de elegante, simple en su forma de expresar código.

Referencias

Programming in Scala, Third Edition Resources

(If you don't already have it, you can purchase Programming in Scala, Third Edition here.)

booksites.artima.com

Pattern Matching | Scala Documentation

Pattern matching is a mechanism for checking a value against a pattern. A successful match can also deconstruct a value…

docs.scala-lang.org

Formatear código en IntelliJ

noreply@blogger.com (fglopez) — Sat, 29 Jul 2023 16:21:00 +0000

SHIFT + ENTER = insertar nueva línea

TAB = escribir palabra clave (por ejemplo, main) y pulsar para sacar live template

CTRL + E = Ver ficheros recientes

CTRL + ALT + L = Reformatear código (resaltado o todo el fichero)

PREFERENCES = Definir los estándares esperados en Editor / Code Style. Buscando Align en la pestaña de Wrapping y localizar Chained Method Calls para poder alineas las llamadas múltiples en diferentes líneas de manera que el punto esté alineado (Align when multiline). Con CTRL+ ALT + L tienen efecto todas estas configuraciones de estilo de código.

ALT + ENTER = Resaltar código y usar estas teclas para mostrar las acciones de contexto y cambiar el formato; se puede cambiar el wrapping de líneas muy largas con estas acciones de contexto.

CTRL + ALT + SHIFT + L = Reformatear todo el código del fichero (permite reformatear solo líneas de codigo modificadas o actuar sobre todo el fichero). Se puede reorganizar código

EditorConfig File = Es un opción que permite crear este fichero para definir las reglas de estilo de código. Inicialmente el fichero contiene la misma configuración del IDE. Sobrescribe los settings del IDE, aunque esta posibilidad se puede desactivar. Se pueden crear diferentes ficheros de estilo de código por cada directorio, aunque esto no es buena práctica. Las entradas de configuracion duplicadas en un directorio se pueden eliminar si queremos que el estilo se herede del fichero padre.

En la barra de estado en la parte inferior derecha podemos abrir el fichero de configuración de estilo que estamos utilizando.

CTRL + Space = Abre sugerencias para completar código para editar de propiedades de estilo

CTRL + SHIFT + A = Find Action en una clase y buscar por rearrange para reorganizar el código de una clase de acuerdo a los settings. También se puede pulsar la opción de buscar en todos los sitios con DOBLE- SHIFT y buscar la opción de rearrange code para reorganizar código.

Poner los métodos privados al final de los públicos = Abrir Preferencias de Editor / Code Style. En el tab de Arrangement definir las reglas a aplicar.

Recursos Data Engineer

noreply@blogger.com (fglopez) — Mon, 24 Jul 2023 18:31:00 +0000

---------------------- ARTICULOS SPARK SCALA ---------------------- https://sparkbyexamples.com/spark/spark-setup-run-with-scala-intellij/

https://medium.com/@Sushil_Kumar/setting-up-spark-with-scala-development-environment-using-intellij-idea-b22644f73ef1

Instalar Java 8 en Ubuntu

- https://computingforgeeks.com/how-to-install-java-8-on-ubuntu/

Instalar intellij-idea en Ubuntu 20.04 LTS

- https://youtu.be/zCkozcDBJ-A

Instalar plugin Scala y crear Proyecto en IntelliJ

https://docs.scala-lang.org/getting-started/intellij-track/getting-started-with-scala-in-intellij.html

Mi primera aplicacion Apache Spark en Scala con Maven en IntelliJ

- https://www.diegocalvo.es/mi-primera-aplicacion-apache-spark-en-scala-con-maven-en-intelligent-idea/

Create Scala Project with Maven in IntelliJ IDEA

https://javahelps.com/create-scala-project-with-maven-in-intellij-idea?amp=1

Spark Scala con Maven en IntelliJ

- https://josedeveloper.com/2020/03/23/spark-scala-con-maven-en-intellij/

Configurar Scala SDK Modificar fichero pom.xml con version de scala y dependencias spark - https://github.com/martinprobson/Spark-Scala-Maven-Example/blob/master/pom.xml

Scala con Maven

- https://docs.scala-lang.org/tutorials/scala-with-maven.html#the-scala-maven-plugin

Scala on IntelliJ using Maven

https://mxnet.apache.org/versions/1.7/api/scala/docs/tutorials/mxnet_scala_on_intellij

Crear App Spark usando el IDE de IntelliJ en Ubuntu

- https://youtu.be/AP5JvdgiBME

Usar Gradle para crear una aplicación simple en Spark Scala

- https://medium.com/@yanggao1119/using-gradle-to-create-a-simple-scala-spark-application-d5f1740e741b

Crear de manera rapida un proyecto scala usando gradle en intellij

- http://www.legendu.net/misc/blog/scala-with-gradle-in-intellij/

Iniciarse en Gradle con IntelliJ

- https://www.jetbrains.com/help/idea/getting-started-with-gradle.html

Generar un fichero JAR de una aplicación Scala con SBT

- https://alvinalexander.com/scala/sbt-how-build-single-executable-jar-file-assembly/

Como configurar y estructurar una aplicación Spark en Scala

- https://godatadriven.com/blog/how-to-setup-and-structure-a-spark-application-in-scala/

How to create maven project for Spark and Scala in Scala IDE - https://medium.com/@manojkumardhakad/how-to-create-maven-project-for-spark-and-scala-in-scala-ide-1a97ac003883

Create your first scala project with Maven - https://docs.scala-lang.org/tutorials/scala-with-maven.html

Create scala project with maven in IntelliJ Idea - https://www.javahelps.com/2021/03/create-scala-with-maven-project-in.html

Setting up a Spark development environment with Scala

https://www.cloudera.com/tutorials/setting-up-a-spark-development-environment-with-scala.html

Instalar Apache Spark y Scala

- https://sundog-spark.s3.amazonaws.com/spark-install.pdf

Setup Development Environment - Scala and Spark

https://kaizen.itversity.com/courses/aws-emr-and-spark-2-using-scala/lessons/setup-development-environment-for-scala-and-spark/topic/setup-development-environment-scala-and-spark/

Creating a Spark Standalone Cluster with Docker and Docker Compose

https://dev.to/mvillarrealb/creating-a-spark-standalone-cluster-with-docker-and-docker-compose-2021-update-6l4

Learning Scala on MacBook with MacOS 12 (Monterey)

https://janrock.medium.com/learning-scala-and-akka-on-new-macbook-with-macos-12-monterey-f516193f1b5e

Setup homebrew on macOS Monterey

https://tech-cookbook.com/2021/10/25/how-to-setup-homebrew-brew-install-on-macos-12-monterey/

Install Java on Mac with brew

https://adamtheautomator.com/install-java-on-mac/

Setup Scala Spark project in IntelliJ using SBT build tool

https://medium.com/@Sushil_Kumar/setting-up-spark-with-scala-development-environment-using-intellij-idea-b22644f73ef1

Kernel de Scala para Jupyter

- https://almond.sh/docs/intro

- https://github.com/almond-sh/examples

Instalar Spark en Ubuntu

- https://phoenixnap.com/kb/install-spark-on-ubuntu

Spark setup with Scala and Run in IntelliJ

https://sparkbyexamples.com/spark/spark-setup-run-with-scala-intellij/?expand_article=1

——— LEER CONFIGURACIONES ——-

Configuration Library for JVM Languages - https://github.com/lightbend/config

Reading configurations In Scala - https://medium.com/@ramkarnani24/reading-configurations-in-scala-f987f839f54d

TypeSafe Config Features and Example Usage - https://www.stubbornjava.com/posts/typesafe-config-features-and-example-usage

Hoy to use the Lightbend Config library In a Scala of Java application - https://alvinalexander.com/scala/how-use-lightbend-config-library-scala-java-application/

Configuring Spark Applications with Typesafe Config - https://florentfo.rest/2019/01/07/configuring-spark-applications-with-typesafe-config.html

----- IDEs --—

Working with Scala on IntelliJ

- https://www.jetbrains.com/help/idea/discover-intellij-idea-for-scala.html

Compile and build applications with IntelliJ

- https://www.jetbrains.com/help/idea/compiling-applications.html

Write and edit source code

- https://www.jetbrains.com/help/idea/working-with-source-code.html

————- JAVA —————————————————————-

Java full course for beginners (using IntelliJ) - https://youtu.be/Qgl81fPcLc8

----------------------------- PROGRAMA DE FORMACION DEVOPS -----------------------------

Git, GitHub and Gitlab - Curso de Atlassian en Coursera: Version Control with Git CI/CD Pipeline with JEnkins - Cursos Udemy: - Jenkins, from zero to hero. Become a DevOps Jenkins Master - GitLab CI: Pipelines, CI/CD and DevOps for Beginners Configuration Management with Chef, Puppet and Ansible - Ansible for the absolute beginner - Hands-on - DevOps - Curso Udemy: Dive into Ansible - From begginer to expert - Todos los cursos en kodekloud.com Docker Certified Associate Certification - Todos los cursos en kodekloud.com - Instalar Docker en Ubuntu: https://docs.docker.com/engine/install/ubuntu/ - Curso Docker y Devops: de novato a experto: https://www.udemy.com/share/102RT4BUQdeF5UQX4=/ - Desplegar una aplicación NodeJS con Docker: https://www.luigicolella.it/blog/how-to-create-nodejs-app-with-docker-1 - Dockerizar una App Scala: https://medium.com/@ievstrygul/dockerizing-scala-app-3fdf08cffda4 - dockerizar una aplicación Django - https://platzi.com/blog/django-docker/ - Como crear imagenes docker para Django - https://www.section.io/engineering-education/django-docker/ - Spark cluster with docker and docker-compose: https://github.com/fgarcialopez605/spark-essentials/tree/master/spark-cluster Certified Kubernetes Administrator - Todos los cursos en kodekloud.com Devops on AWS - Curso Udemy: AWS Certified DevOps Engineer (Requiere al menos el AWS Certified Developer)

Apache Spark Cluster on Docker:

- https://www.kdnuggets.com/2020/07/apache-spark-cluster-docker.html

Spark clúster with docker

https://github.com/mvillarrealb/docker-spark-cluster

Apache Spark Standalone Cluster on Docker

https://github.com/cluster-apps-on-docker/spark-standalone-cluster-on-docker#tech-stack

---------------------------- PROGRAMA CERTIFICACION CLOUD ----------------------------

https://acloudguru.com/

AWS Cloud Practitioner Certification

- https://www.whizlabs.com/aws-certified-cloud-practitioner/

AWS Certified Cloud Practitioner

https://www.udemy.com/share/103a09BUQdeF5UQX4=/

Official Google Cloud Certified Professional Data Engineer Study Guide

https://www.amazon.es/Official-Google-Certified-Professional-Engineer/dp/1119618436/ref=sr_1_1?__mk_es_ES=%C3%85M%C3%85%C5%BD%C3%95%C3%91&dchild=1&keywords=Official+Google+Cloud+Certified+Professional+Data+Engineer+Study+Guide&qid=1620994245&sr=8-1

AWS Certified Solutions Architect

https://www.udemy.com/course/aws-certified-solutions-architect-associate-saa-c02/

AWS Certified Solutions Architect Study Guide

https://www.amazon.com/Certified-Solutions-Architect-Study-Guide-dp-111950421X/dp/111950421X?tag=javamysqlanta-20

Prepare for AWS Solution Architect Associate Certification

https://medium.com/javarevisited/how-to-prepare-for-aws-solution-architect-associate-certification-saa-c01-saa-c02-exam-in-2021-a6e7e7e771fc

GCP Professional Data Engineer Dmitri Lerko - https://deploy.live/blog/google-cloud-certified-professional-data-engineer/

Como superar el examen de data Engineer de Google cloud - https://dbibyhavas.io/es/blog/data-engineer-de-google-cloud/

Desarrollo Python en Google Cloud

- https://cloud.google.com/python/docs/setup

- https://cloud.google.com/python

Desarrollo Java en Google Cloud

- https://cloud.google.com/java/docs/setup

- https://cloud.google.com/java

—————————- PROGRAMA CERTIFICACIÓN SPARK ——————————-

Como aprobar la certificación Databricks Spark 3.0

- http://josedeveloper.com/2021/04/30/como-aprobar-la-certificacion-databricks-spark-3-0/

------------------- MAQUINAS VIRTUALES -------------------

Install Ubuntu 20.04 LTS on VirtualBox in Windows 10 - https://youtu.be/x5MhydijWmc - https://gadgetliv.com/install-ubuntu-on-virtualbox-in-windows/

How to install Ubuntu in VirtualBox on Windows 10

- https://youtu.be/IOwlnpWPuj0

Como empezar a trabajar con un contenedor Docker de Cloudera QuickStart

https://medium.com/@SnazzyHam/how-to-get-up-and-running-with-clouderas-quickstart-docker-container-732c04ed0280

Proyecto Vagrant para arrancar una máquina virtual con Hadoop/Spark

https://github.com/martinprobson/vagrant-hadoop-hive-spark

-------------------- DESARROLLO PYTHON --------------------

Jupyter Notebooks, JupyterLab, Zeppelin, Anaconda, Spyder

Iniciar un proyecto de ciencia de datos con Anaconda en Python - http://spaceanalytics.blogspot.com/2020/11/iniciar-un-proyecto-de-ciencia-de-datos.html

Empezar a trabajar con Apache Spark y Python

- https://www.sundog-education.com/spark-python/

Introducción práctica al analisis de datos en python

- https://towardsdatascience.com/a-beginners-guide-to-data-analysis-in-python-188706df5447

Instalar un entorno de desarrollo Python3 en Ubuntu 20.04

- https://www.digitalocean.com/community/tutorials/how-to-install-python-3-and-set-up-a-programming-environment-on-an-ubuntu-20-04-server

Creación de proyectos Python en IntelliJ

- https://www.jetbrains.com/help/idea/plugin-overview.html

Introducción a MongoDB, PySpark y Jupyter Notebook

- https://www.mongodb.com/blog/post/getting-started-with-mongodb-pyspark-and-jupyter-notebook

Setup PySpark environment on Ubuntu

- https://medium.com/@GalarnykMichael/install-spark-on-ubuntu-pyspark-231c45677de0

Setup JupyterLab Environment on Ubuntu

https://www.digitalocean.com/community/tutorials/how-to-set-up-a-jupyterlab-environment-on-ubuntu-18-04

Install PySpark on Ubuntu

https://t.co/UdzmhAaTKi

https://www.roseindia.net/bigdata/pyspark/install-pyspark-on-ubuntu.shtml

Install PySpark locally and use it with Jupyter Notebook

- https://youtu.be/MLXOy-OhWRY

PySpark crash course in 50 minutes

- https://youtu.be/6rgoqfzX_YI

PySpark Tutorial for Beginners

https://sparkbyexamples.com/pyspark-tutorial/

Instalar Anaconda Python en Ubuntu 20.04

https://www.digitalocean.com/community/tutorials/how-to-install-the-anaconda-python-distribution-on-ubuntu-20-04-es

Documentación oficial de Anaconda

- https://docs.anaconda.com/anaconda/

JupyterLab in a virtual environment

- https://dev.to/joeneville_/jupyterlab-in-a-virtual-environment-gf5

Use Virtual Environments Inside Jupyter Notebooks & JupyterLab

- https://www.zainrizvi.io/blog/jupyter-notebooks-best-practices-use-virtual-environments/

Create Django App on Ubuntu Server

- https://linuxhint.com/create_django_app_ubuntu/

Documentación oficial de Conda

https://conda.io/projects/conda/en/latest/user-guide/getting-started.html

Como instalar Python 3 y entornos virtuales con miniconda en Ubuntu

https://thucnc.medium.com/how-to-install-python-3-and-virtual-environments-with-miniconda-in-ubuntu-18-04-c9e000a39994

Gestionar múltiples versiones de Python en Ubuntu

https://hackersandslackers.com/multiple-versions-python-ubuntu/

Otros ----- http://www.blogdelujo.com/2010/01/ver-incrustar-otra-web-en-blog.html

—————————DESARROLLO KAFKA ——————-

Desarrollo de aplicaciones Kafka con Scala -

https://www.confluent.io/blog/kafka-scala-tutorial-for-beginners/

Kafka Streams DSL en Scala - https://medium.com/@danieljameskay/kafka-streams-dsl-for-scala-the-basics-11d603295f5c

———— KUDU / IMPALA —————

Montar una infraestructura Docker que usa Impala para operar con Kudu

https://github.com/apache/kudu/tree/master/examples/quickstart/impala

Oficial Kudu - https://kudu.apache.org/

Oficial Impala - https://impala.apache.org/index.html

Tutoríales prácticos de Impala - https://impala.apache.org/docs/build/html/topics/impala_tutorial.html

Building real time BI systems with Kafka, Spark and Kudu - https://databricks.com/session/building-real-time-bi-systems-with-kafka-spark-and-kudu

Apache Kudu and Spark SQL for fast Analytics on fast data - https://databricks.com/session/apache-kudu-and-spark-sql-for-fast-analytics-on-fast-data

CERTIFICACIÓN GCP DATA ENGINEER

https://www.cloudskillsboost.google/paths/16

https://youtu.be/w2_ukhoLGEM

Usar Google Dataproc con PySpark y Jupyter Notebook

https://morioh.com/p/2c1e7be91f53

Run PySpark job on Dataproc

https://acloudguru.com/hands-on-labs/running-a-pyspark-job-on-cloud-dataproc-using-google-cloud-storage

GENERAL DATA ENGINEER

https://awesomedataengineering.com/

https://medium.com/coriers/data-engineering-roadmap-for-2021-eac7898f0641

Setting up a Dev Environment for Data Engineers

- https://discuss.itversity.com/t/data-engineers-setting-up-the-development-environment/19879

Data Architecture Articles

https://medium.com/tag/data-architecture

How to become a Data Architect

https://techguide.org/careers/data-architect/

Top Data Architect and Data Engineer Certifications

https://www.techtarget.com/whatis/feature/8-top-data-architect-and-data-engineer-certifications

Transformar dataframe spark

noreply@blogger.com (fglopez) — Wed, 12 Jul 2023 15:30:00 +0000

def transformArrayStruct(df: DataFrame): DataFrame = {

df.withColumn("array_of_struct", expr("transform(array_of_struct, x -> struct(x.a, x.c))")) }

import org.apache.spark.sql.functions.{col, isArray, isStruct} import org.apache.spark.sql.{DataFrame, SparkSession} def isColumnArrayStruct(df: DataFrame, columnName: String): Boolean = { val arrayCheck = isArray(col(columnName)) val structCheck = isStruct(element_at(col(columnName), 1)) df.select(arrayCheck && structCheck).as[Boolean].head() }

Extraer partes del nombre de una entidad MySQL

noreply@blogger.com (fglopez) — Thu, 06 Jul 2023 08:53:00 +0000

Problema: Tengo un String que es una dirección mysql de una tabla y quiero hacer una función scala que devuelva una lista con 2 elementos, el primero con la parte izquierda de la dirección mysql hasta el último carácter / (sin incluirlo) y el segundo con la parte derecha que hay después del carácter /, que corresponde a la base de datos.

Puedes lograr esto en Scala utilizando métodos de manipulación de cadenas. Aquí tienes una función que toma una dirección MySQL y devuelve una lista con la parte izquierda y derecha de la dirección:

def splitMySQLAddress(address: String): List[String] = { val lastIndex = address.lastIndexOf("/") val leftPart = address.substring(0, lastIndex) val rightPart = address.substring(lastIndex + 1) List(leftPart, rightPart) }

Aquí tienes un ejemplo de cómo puedes usar esta función:

val mysqlAddress = "jdbc:mysql://localhost:3306/mydatabase" val parts = splitMySQLAddress(mysqlAddress) println(parts) // Imprime: List(jdbc:mysql://localhost:3306, mydatabase)

En este ejemplo, la función splitMySQLAddress toma la dirección MySQL como entrada y utiliza el método lastIndexOf para encontrar la posición del último carácter "/". Luego, utiliza el método substring para dividir la dirección en dos partes, la parte izquierda y la parte derecha. Estas partes se agregan a una lista y se devuelve como resultado.

Ten en cuenta que esta implementación asume que la dirección MySQL siempre tiene el formato esperado, es decir, tiene al menos un carácter "/" y no contiene más de un carácter "/" al final de la parte izquierda. Si necesitas manejar casos especiales, como direcciones incorrectas o formatos diferentes, puedes ajustar la función según tus necesidades.

Comparación de esquema entre Parquet y Athena

noreply@blogger.com (fglopez) — Thu, 22 Jun 2023 11:10:00 +0000

Versión 1:

from pyspark.sql import SparkSession

# Crear la sesión de Spark

spark = SparkSession.builder \

.appName("Comparar Esquema de Parquet y Tabla en Athena") \

.config("hive.metastore.glue.catalog.id", "glue_catalog_id") \

.enableHiveSupport() \

.getOrCreate()

# Nombre de la base de datos y tabla en Athena

database_name = "nombre_base_de_datos"

table_name = "nombre_tabla"

# Ruta del archivo Snappy Parquet

parquet_file_path = "ruta_del_archivo.parquet"

# Leer el esquema del archivo Snappy Parquet

parquet_df = spark.read.format("parquet").load(parquet_file_path)

parquet_schema = parquet_df.schema

# Leer el esquema de la tabla en Athena

athena_df = spark.table(f"{database_name}.{table_name}")

athena_schema = athena_df.schema

# Comparar los esquemas

if parquet_schema == athena_schema:

print("El esquema del archivo Snappy Parquet y la tabla en Athena son iguales.")

else:

print("El esquema del archivo Snappy Parquet y la tabla en Athena son diferentes.")

# Cerrar la sesión de Spark

spark.stop()

Una mejora al código anterior para que imprima las diferencias entre el esquema del archivo Snappy Parquet y el esquema de la tabla en Athena:

Una mejora adicional al programa anterior para que también imprima las diferencias en el tipo de columna entre el esquema del archivo Snappy Parquet y el esquema de la tabla en Athena:

from pyspark.sql import SparkSession # Crear la sesión de Spark spark = SparkSession.builder \ .appName("Comparar Esquema de Parquet y Tabla en Athena") \ .config("hive.metastore.glue.catalog.id", "glue_catalog_id") \ .enableHiveSupport() \ .getOrCreate() # Nombre de la base de datos y tabla en Athena database_name = "nombre_base_de_datos" table_name = "nombre_tabla" # Ruta del archivo Snappy Parquet parquet_file_path = "ruta_del_archivo.parquet" # Leer el esquema del archivo Snappy Parquet parquet_df = spark.read.format("parquet").load(parquet_file_path) parquet_schema = parquet_df.schema # Leer el esquema de la tabla en Athena athena_df = spark.table(f"{database_name}.{table_name}") athena_schema = athena_df.schema # Obtener las columnas presentes en el esquema del archivo Snappy Parquet pero no en el esquema de la tabla en Athena missing_columns_in_athena = set(parquet_schema.fieldNames()) - set(athena_schema.fieldNames()) # Obtener las columnas presentes en el esquema de la tabla en Athena pero no en el esquema del archivo Snappy Parquet missing_columns_in_parquet = set(athena_schema.fieldNames()) - set(parquet_schema.fieldNames()) # Obtener las columnas con tipos diferentes entre el esquema del archivo Snappy Parquet y el esquema de la tabla en Athena different_type_columns = [] for field in parquet_schema.fields: if field.name in athena_schema: athena_field = athena_schema[field.name] if field.dataType != athena_field.dataType: different_type_columns.append((field.name, field.dataType, athena_field.dataType)) if missing_columns_in_athena: print("Columnas presentes en el archivo Snappy Parquet pero no en la tabla de Athena:") for column in missing_columns_in_athena: print(column) if missing_columns_in_parquet: print("Columnas presentes en la tabla de Athena pero no en el archivo Snappy Parquet:") for column in missing_columns_in_parquet: print(column) if different_type_columns: print("Columnas con tipos diferentes entre el archivo Snappy Parquet y la tabla de Athena:") for column, parquet_type, athena_type in different_type_columns: print(f"Columna: {column}, Tipo en Parquet: {parquet_type}, Tipo en Athena: {athena_type}") # Cerrar la sesión de Spark spark.stop()

Prueba de lectura de Parameter Store en ScalaTest

noreply@blogger.com (fglopez) — Mon, 19 Jun 2023 12:09:00 +0000

import org.scalatest.flatspec.AnyFlatSpec

import org.scalatest.matchers.should.Matchers

import org.scalatestplus.mockito.MockitoSugar

import software.amazon.awssdk.services.ssm.SsmClient

import software.amazon.awssdk.services.ssm.model.{GetParameterRequest, GetParameterResponse, Parameter}

import scala.util.{Success, Try}

trait ParameterStoreReader {

protected val ssmClient: SsmClient

def getParameter(parameterName: String): Try[String] = {

val request = GetParameterRequest.builder().name(parameterName).build()

val response = ssmClient.getParameter(request)

Success(response.parameter().value())

}

class ParameterStoreReaderSpec extends AnyFlatSpec with Matchers with MockitoSugar {

it should "retrieve a parameter from AWS Parameter Store" in {

val expectedParameterValue = "parameterValue"

// Create a mock SsmClient

val ssmClientMock: SsmClient = mock[SsmClient]

// Create an instance of the trait to be tested, passing the mock SsmClient

val parameterStoreReader: ParameterStoreReader = new ParameterStoreReader {

override protected val ssmClient: SsmClient = ssmClientMock

}

// Simulate the response from AWS Parameter Store

val response = GetParameterResponse.builder()

.parameter(Parameter.builder()

.name("parameterName")

.value(expectedParameterValue)

.build())

.build()

// Stub the getParameter method of the mock SsmClient

when(ssmClientMock.getParameter(any[GetParameterRequest])).thenReturn(response)

// Test the getParameter method

val result = parameterStoreReader.getParameter("parameterName")

result shouldBe Success(expectedParameterValue)

// Verify that the getParameter method was called with the correct parameter name

verify(ssmClientMock).getParameter(GetParameterRequest.builder().name("parameterName").build())

}

Pruebas unitarias de llamada a APIs en ScalaTest

noreply@blogger.com (fglopez) — Fri, 16 Jun 2023 21:31:00 +0000

import org.scalatest.flatspec.AnyFlatSpec

import org.scalatest.matchers.should.Matchers

import requests.Response

import scala.collection.mutable

object DatahubGraphqlProcessor {

def sendRequest(url: String): Response = {

// Realizar la petición a la API y obtener la respuesta

// ...

// Simulamos una respuesta ficticia para el ejemplo

val mockResponseCode: Int = 200

val mockResponseText: String = """{"message": "Mock response"}"""

Response(

statusCode = mockResponseCode,

headers = Map.empty,

body = mockResponseText

)

}

def parseResponse(responseText: String): mutable.Map[String, List[String]] = {

// Parsear el texto JSON y construir el mapa mutable

// ...

// Simulamos un resultado ficticio para el ejemplo

val parsedResult: mutable.Map[String, List[String]] = mutable.Map(

"key1" -> List("value1", "value2"),

"key2" -> List("value3", "value4")

)

parsedResult

}

class DatahubGraphqlProcessorSpec extends AnyFlatSpec with Matchers {

val mockResponseCode: Int = 200

val mockResponseText: String = """{"message": "Mock response"}"""

"DatahubGraphqlProcessor" should "simular la llamada a la API REST y parsear la respuesta" in {

// Mockear la llamada a la API

val mockDatahubGraphqlProcessor = new DatahubGraphqlProcessor {

override def sendRequest(url: String): Response = {

// Simular la respuesta de la API

Response(

statusCode = mockResponseCode,

headers = Map.empty,

body = mockResponseText

)

}

// Realizar la llamada y verificar la respuesta

val response: Response = mockDatahubGraphqlProcessor.sendRequest("https://api.example.com/graphql")

response.statusCode shouldEqual mockResponseCode

response.text() shouldEqual mockResponseText

// Parsear la respuesta

val parsedResult: mutable.Map[String, List[String]] = mockDatahubGraphqlProcessor.parseResponse(response.text())

// Verificar el resultado del parsing

parsedResult("key1") shouldEqual List("value1", "value2")

parsedResult("key2") shouldEqual List("value3", "value4")

}

Pruebas unitarias de tipo mock en ScalaTest (SsmUtils)

noreply@blogger.com (fglopez) — Fri, 16 Jun 2023 20:31:00 +0000

No es posible simular la instanciación de un cliente SsmClient con un Parameter Store en local para pruebas unitarias utilizando credenciales anónimas. La biblioteca AWS SDK requiere autenticación válida para acceder a los servicios de AWS, incluido Parameter Store.

En las pruebas unitarias, es recomendable utilizar enfoques de simulación o mocking para simular el comportamiento de las llamadas a servicios externos, en lugar de intentar ejecutarlos localmente. Puedes utilizar bibliotecas de simulación como Mockito, MockServer u otros frameworks similares para simular las respuestas de las llamadas a los servicios de AWS, incluyendo la respuesta del Parameter Store.

En lugar de interactuar con el Parameter Store real durante las pruebas unitarias, puedes crear mocks o stubs para simular las respuestas del Parameter Store y probar el comportamiento de tu código en función de esas respuestas simuladas. Esto te permitirá aislar tus pruebas y hacerlas más rápidas, predecibles y menos dependientes de recursos externos como el Parameter Store de AWS.

Recuerda que las pruebas unitarias se centran en probar la lógica interna de tu código de manera aislada, sin depender de servicios externos. Por lo tanto, es preferible simular o simular parcialmente los servicios externos para garantizar pruebas más confiables y eficientes.

import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers import software.amazon.awssdk.services.ssm.SsmClient import software.amazon.awssdk.services.ssm.model.{GetParameterRequest, GetParameterResponse} trait MyParameterStore { implicit val ssmClient: SsmClient def getParameterValue(parameterName: String): String = { val request = GetParameterRequest.builder() .name(parameterName) .build() val response = ssmClient.getParameter(request) response.parameter().value() } } class MyParameterStoreSpec extends AnyFlatSpec with Matchers with MyParameterStore { // Cliente Ssm simulado para las pruebas implicit val ssmClient: SsmClient = new SsmClient { override def getParameter(request: GetParameterRequest): GetParameterResponse = { request.name() match { case "parameter1" => GetParameterResponse.builder() .parameter(Parameter.builder() .name(request.name()) .value("value1") .build()) .build() case "parameter2" => GetParameterResponse.builder() .parameter(Parameter.builder() .name(request.name()) .value("value2") .build()) .build() case _ => throw new IllegalArgumentException(s"Unknown parameter: ${request.name()}") } } } "MyParameterStore" should "obtener un parámetro del Parameter Store" in { val parameterValue1 = getParameterValue("parameter1") parameterValue1 should be("value1") val parameterValue2 = getParameterValue("parameter2") parameterValue2 should be("value2") } }

En este ejemplo, el trait MyParameterStore define un valor implícito ssmClient de tipo SsmClient. La clase de prueba MyParameterStoreSpec extiende AnyFlatSpec, Matchers y MyParameterStore. Al mezclar MyParameterStore en la clase de prueba, se puede sobrescribir el valor implícito ssmClient para utilizar un cliente Ssm simulado en lugar de uno real.

Dentro del cliente Ssm simulado, se utiliza una coincidencia de patrones (match) para determinar qué parámetro se está solicitando y devolver una respuesta simulada correspondiente. En el ejemplo, se simulan dos parámetros diferentes: "parameter1" y "parameter2", con sus respectivos valores "value1" y "value2".

En el caso de prueba, se llama al método getParameterValue de MyParameterStore y se verifica que los valores de los parámetros sean los esperados.

Recuerda adaptar este código a tus necesidades específicas y utilizar los nombres de parámetros y valores correctos para tu caso de uso.

Generar un ZIP de objetos S3 con un prefijo de clave comun

noreply@blogger.com (fglopez) — Wed, 14 Jun 2023 12:11:00 +0000

En este código, el trait S3ZipManager se encarga de las responsabilidades relacionadas con la generación del archivo ZIP y la subida a S3. El objeto S3ZipGenerator extiende este trait y proporciona la implementación de los valores bucketName y s3KeyPrefix. Luego, en el método main, simplemente llamamos al método generateAndUploadZip() para generar y subir el archivo ZIP.

hemos extraído la lógica repetitiva a métodos privados. El método createS3Client() se encarga de crear el cliente S3, generateZipBytes() genera los bytes del archivo ZIP, listObjectsInS3() lista los objetos en S3, addToZip() agrega un archivo al ZIP y uploadZipToS3() sube el archivo ZIP a S3.

import java.io.{ByteArrayInputStream, ByteArrayOutputStream}

import java.util.zip.{ZipEntry, ZipOutputStream}

import software.amazon.awssdk.auth.credentials.{DefaultCredentialsProvider, StaticCredentialsProvider}

import software.amazon.awssdk.regions.Region

import software.amazon.awssdk.services.s3.{S3Client, S3Configuration}

import software.amazon.awssdk.services.s3.model.{CreateBucketRequest, PutObjectRequest}

import scala.collection.JavaConverters._

import java.time.LocalDate

import java.time.format.DateTimeFormatter

trait S3ZipManager {

def bucketName: String

def s3KeyPrefix: String

def generateAndUploadZip(): Unit = {

val zipFileName = s"your-prefix_${LocalDate.now().format(DateTimeFormatter.ofPattern("yyyyMMdd"))}.zip"

val s3Client = createS3Client()

val zipFileBytes = generateZipBytes(s3Client)

uploadZipToS3(s3Client, zipFileName, zipFileBytes)

println(s"ZIP file $zipFileName created and uploaded to S3.")

}

private def createS3Client(): S3Client = {

S3Client.builder()

.region(Region.US_EAST_1) // Specify your desired region

.credentialsProvider(DefaultCredentialsProvider.create())

.serviceConfiguration(S3Configuration.builder().pathStyleAccessEnabled(true).build())

.build()

}

private def generateZipBytes(s3Client: S3Client): Array[Byte] = {

val byteArrayOutputStream = new ByteArrayOutputStream()

val zipOutputStream = new ZipOutputStream(byteArrayOutputStream)

listObjectsInS3(s3Client).foreach { s3Object =>

val s3ObjectKey = s3Object.key()

val s3ObjectContent = s3Client.getObject(builder =>

builder.bucket(bucketName).key(s3ObjectKey)

).readAllBytes()

addToZip(zipOutputStream, s3ObjectKey, s3ObjectContent)

}

zipOutputStream.close()

byteArrayOutputStream.toByteArray

}

private def listObjectsInS3(s3Client: S3Client): Seq[software.amazon.awssdk.services.s3.model.S3Object] = {

val listObjectsResponse = s3Client.listObjects(builder =>

builder.bucket(bucketName).prefix(s3KeyPrefix)

)

listObjectsResponse.contents().asScala.toSeq

}

private def addToZip(zipOutputStream: ZipOutputStream, entryName: String, content: Array[Byte]): Unit = {

zipOutputStream.putNextEntry(new ZipEntry(entryName))

zipOutputStream.write(content)

zipOutputStream.closeEntry()

}

private def uploadZipToS3(s3Client: S3Client, zipFileName: String, zipFileBytes: Array[Byte]): Unit = {

val zipFileInputStream = new ByteArrayInputStream(zipFileBytes)

val putObjectRequest = PutObjectRequest.builder()

.bucket(bucketName)

.key(zipFileName)

.contentType("application/zip")

.build()

s3Client.putObject(putObjectRequest, zipFileInputStream)

}

object S3ZipGenerator extends S3ZipManager {

override val bucketName: String = "your-bucket-name"

override val s3KeyPrefix: String = "your/s3/key/prefix/"

def main(args: Array[String]): Unit = {

generateAndUploadZip()

}

Procesamiento de peticiones GraphQL en Datahub

noreply@blogger.com (fglopez) — Sat, 10 Jun 2023 22:22:00 +0000

A continuación se implementa un código en Scala con 2 objetos singleton. El primero de ellos se llamará DatahubExportRunner que recibe en el constructor la configuración de la aplicacion y se encarga de llamar a otro objeto singleton DatahubGraphqlProcessor pasando un prefijo de clave S3 en el constructor. En función del estado obtenido en la respuesta, que debe ser una instancia de case class con el estado del procesamiento de la petición y un mapa con clave correspondiente a la clave S3 y el valor con una lista de String con el contenido a guardar para ese objeto S3, el objeto DatahubExportRunner escribe el contenido en S3 o lanza una excepción indicando que no se ha podido realizar la exportación por estado incorrecto devuelto al hacer la petición al endpoint graphql de datahub. El objeto DatahubGraphqlProcessor se encarga de hacer una petición http al api rest de Datahub usando el endpoint de graphql, procesar la respuesta y devolver una instancia indicando el resultado.

El método runExport del objeto DatahubExportRunner recibe la configuración como parámetro y obtiene el s3KeyPrefix a partir de dicha configuración. Luego se pasa el prefijo de clave S3 al método que procesa la respuesta, ya que se usa para generar el mapa con el contenido a guardar en S3.

import scala.collection.mutable.ListBuffer import scala.util.{Failure, Success, Try} import scalaj.http.{Http, HttpResponse} // Case class para representar el resultado de la petición graphql case class DatahubGraphqlResponse(status: String, s3Data: Map[String, List[String]]) // Objeto singleton DatahubGraphqlProcessor para hacer la petición graphql y procesar la respuesta object DatahubGraphqlProcessor { def processGraphqlRequest(endpoint: String): DatahubGraphqlResponse = { // Realizar la petición HTTP al endpoint de graphql de Datahub val response: HttpResponse[String] = Http(endpoint).asString // Procesar la respuesta y devolver una instancia de DatahubGraphqlResponse if (response.is2xx) { // Parsear la respuesta y obtener los datos de S3 val s3Data: Map[String, List[String]] = parseResponse(response.body) // Crear la instancia de DatahubGraphqlResponse con el estado "Success" y los datos de S3 DatahubGraphqlResponse("Success", s3Data) } else { // Devolver una instancia de DatahubGraphqlResponse con el estado "Failure" y un mapa vacío DatahubGraphqlResponse("Failure", Map.empty) } } private def parseResponse(responseBody: String): Map[String, List[String]] = { // Implementar el parsing de la respuesta y devolver los datos de S3 en un mapa // ... Map.empty[String, List[String]] } } // Objeto singleton DatahubExportRunner para ejecutar la exportación de datos object DatahubExportRunner { def runExport(configuration: AppConfig): Unit = { // Construir el endpoint de graphql usando la configuración val endpoint: String = buildGraphqlEndpoint(configuration) // Llamar a DatahubGraphqlProcessor para obtener el resultado de la petición graphql val response: DatahubGraphqlResponse = DatahubGraphqlProcessor.processGraphqlRequest(endpoint) // Comprobar el estado de la respuesta y escribir el contenido en S3 o lanzar una excepción response.status match { case "Success" => val s3KeyPrefix = configuration.s3KeyPrefix writeContentToS3(response.s3Data, s3KeyPrefix) println("Data export successful.") case "Failure" => throw new RuntimeException("Data export failed.") } } private def buildGraphqlEndpoint(configuration: AppConfig): String = { // Implementar la construcción del endpoint de graphql usando la configuración // ... "" } private def writeContentToS3(s3Data: Map[String, List[String]], s3KeyPrefix: String): Unit = { // Implementar la lógica para escribir el contenido en S3 usando el prefijo de clave // ... } } // Case class para la configuración de la aplicación case class AppConfig(apiEndpoint: String, apiKey: String, s3KeyPrefix: String /* Otros parámetros */) // Ejemplo de uso val configuration = AppConfig("https://api.datahub.com/graphql", "API_KEY", "my-bucket/data/") DatahubExportRunner.runExport(configuration)

Instanciacion de cliente S3 con SDK de AWS

noreply@blogger.com (fglopez) — Fri, 09 Jun 2023 20:51:00 +0000

Para crear un cliente S3 utilizando los métodos create de ambas SDKs de Java, asumiendo que se ejecuta en AWS con toda la configuración y permisos ya establecidos:

SDK 1 de Java:

import com.amazonaws.services.s3.AmazonS3Client

val s3Client = new AmazonS3Client()

SDK 2 de Java:

import software.amazon.awssdk.services.s3.S3Client

val s3Client: S3Client = S3Client.create()

En la SDK 1 de Java, simplemente creamos una nueva instancia de AmazonS3Client sin pasar ningún argumento al constructor. En este caso, el cliente S3 utilizará automáticamente las credenciales y la configuración establecidas en el entorno de ejecución de AWS.

En la SDK 2 de Java, llamamos al método estático create() de S3Client para crear una instancia del cliente S3. Al igual que en la SDK 1, en este caso también se utilizarán automáticamente las credenciales y la configuración establecidas en el entorno de ejecución de AWS.

Recuerda que en ambos casos, es importante asegurarse de que la instancia se esté ejecutando en un entorno de AWS con los permisos y configuraciones adecuados para acceder al servicio S3.

Generador de ficheros en S3

noreply@blogger.com (fglopez) — Fri, 09 Jun 2023 16:06:00 +0000

hemos utilizado las clases y métodos de la SDK 2 de AWS, como S3Client y PutObjectRequest, para crear un cliente S3 y construir la solicitud de putObject. Además, hemos utilizado RequestBody.fromString para convertir el contenido del fichero a un objeto de solicitud de cuerpo.

En este código, hemos creado un trait llamado S3FileWriter que define el método writeContent. Este método acepta el nombre del bucket, la clave S3 y las líneas del fichero a escribir. Dentro del método, se instancia el cliente S3 y se realiza la llamada a putObject para escribir el contenido en S3.

La clase FileGenerator ahora extiende el trait S3FileWriter y utiliza el método writeContent para escribir el contenido de cada fichero en S3.

De esta manera, el código se ha modularizado y reutilizado mediante el uso de traits en Scala.

hemos creado un objeto singleton FileLinesMap que encapsula la estructura del mapa mutable linesMap. Hemos agregado un método addLine que añade dinámicamente las líneas a medida que aparecen. El método verifica si la clave ya existe en el mapa, y si es así, actualiza la lista de líneas agregando la nueva línea. Si la clave no existe, se agrega una nueva entrada al mapa con la primera línea del fichero en el valor.

La clase FileGenerator ahora utiliza el objeto FileLinesMap para agregar dinámicamente las líneas de los ficheros. Al final, se obtiene el mapa de líneas y se escribe el contenido en S3 utilizando el método writeContent.

import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider import software.amazon.awssdk.core.sync.RequestBody import software.amazon.awssdk.regions.Region import software.amazon.awssdk.services.s3.S3Client import software.amazon.awssdk.services.s3.model.PutObjectRequest import java.nio.charset.StandardCharsets

import scala.collection.mutable.{Map => MutableMap}

object FileLinesMap {

private val linesMap: MutableMap[String, List[String]] = MutableMap.empty

def addLine(s3Key: String, line: String): Unit = {

val updatedLines = linesMap.getOrElse(s3Key, Nil) :+ line

linesMap.put(s3Key, updatedLines)

}

def getLinesMap: MutableMap[String, List[String]] = linesMap

}

trait S3FileWriter {

def writeContent(bucketName: String, s3Key: String, lines: List[String]): Unit = { val s3Client: S3Client = S3Client.builder() .region(Region.US_EAST_1) .credentialsProvider(DefaultCredentialsProvider.create()) .build() val fileContent = lines.mkString("\n") val putObjectRequest = PutObjectRequest.builder() .bucket(bucketName) .key(s3Key) .build() s3Client.putObject(putObjectRequest, RequestBody.fromString(fileContent, StandardCharsets.UTF_8)) }

}

object FileGenerator extends S3FileWriter {

def main(args: Array[String]): Unit = {

val bucketName = "nombre-del-bucket"

val s3Keys = List("ruta/fichero1.csv", "ruta/fichero2.csv", "ruta/fichero3.csv")

// Agrega las líneas generadas al mapa dinámicamente

FileLinesMap.addLine("ruta/fichero1.csv", "Línea 1 del fichero 1")

FileLinesMap.addLine("ruta/fichero1.csv", "Línea 2 del fichero 1")

FileLinesMap.addLine("ruta/fichero2.csv", "Línea 1 del fichero 2")

FileLinesMap.addLine("ruta/fichero2.csv", "Línea 2 del fichero 2")

FileLinesMap.addLine("ruta/fichero2.csv", "Línea 3 del fichero 2")

FileLinesMap.addLine("ruta/fichero3.csv", "Línea 1 del fichero 3")

FileLinesMap.addLine("ruta/fichero3.csv", "Línea 2 del fichero 3")

FileLinesMap.addLine("ruta/fichero3.csv", "Línea 3 del fichero 3")

FileLinesMap.addLine("ruta/fichero3.csv", "Línea 4 del fichero 3")

// Obtiene el mapa de líneas

val fileLinesMap = FileLinesMap.getLinesMap

// Escribe el contenido en Amazon S3 para cada fichero

fileLinesMap.foreach { case (s3Key, lines) =>

writeContent(bucketName, s3Key, lines)

}

Obtener todos los ficheros CSV dentro de un directorio

noreply@blogger.com (fglopez) — Fri, 09 Jun 2023 12:20:00 +0000

def getAllCsvFiles(directory: File): List[File] = {

if (directory.exists() && directory.isDirectory) { val csvFiles = directory.listFiles().flatMap { file => if (file.isFile && file.getName.endsWith(".csv")) { List(file) } else if (file.isDirectory) { getAllCsvFiles(file) } else { List.empty[File] } } csvFiles.toList } else { List.empty[File] } } }

Subida recursiva de archivos CSV en un directorio local a S3

noreply@blogger.com (fglopez) — Thu, 08 Jun 2023 21:06:00 +0000

Vamos a crear un trait CsvUploader que se encarga de la lógica de subir archivos CSV a S3 utilizando la AWS CLI. El método uploadCsvFiles recibe la ruta del directorio local donde se encuentran los archivos CSV, el nombre del bucket en S3 y el prefijo de ruta en S3 para los archivos. Luego, se itera sobre los archivos CSV dentro del directorio y se ejecuta el comando de subida utilizando la AWS CLI.

La clase S3BatchUpload utiliza el trait CsvUploader para subir todos los archivos CSV dentro de un directorio específico a S3. Simplemente se especifica la ruta del directorio local, el nombre del bucket en S3 y el prefijo de ruta en S3, y se llama al método uploadCsvFiles para realizar la subida.

Para que la estructura de directorios se mantenga en el bucket de destino se construye la clave s3Key en función de la estructura de directorios local. Se utiliza substring para obtener la parte de la ruta después del directorio base, y luego se reemplaza el separador de archivos local (File.separator) por el separador de rutas en S3 ("/"). De esta manera, se mantendrá la misma estructura de directorios en el bucket de destino.

import java.io.File

import sys.process._

trait AwsCliExecutor {

protected def executeCommand(command: String): Int = command.!

}

trait CsvUploader extends AwsCliExecutor {

def uploadCsvFiles(directoryPath: String, bucketName: String, s3KeyPrefix: String): Unit = {

val directory = new File(directoryPath)

val csvFiles = directory.listFiles(_.getName.toLowerCase.endsWith(".csv"))

csvFiles.foreach { file =>

val localFilePath = file.getAbsolutePath

val s3Key = s3KeyPrefix + localFilePath.substring(directoryPath.length).replace(File.separator, "/")

val uploadCommand = s"aws s3 cp $localFilePath s3://$bucketName/$s3Key"

val result = executeCommand(uploadCommand)

if (result == 0) {

println(s"Archivo ${file.getName} subido exitosamente a S3 con ruta $s3Key.")

} else {

println(s"Ocurrió un error al subir el archivo ${file.getName} a S3.")

}

object S3BatchUpload extends App with CsvUploader {

val localDirectoryPath = "/path/to/your/local/directory" // Ruta del directorio local con los archivos CSV

val bucketName = "your-bucket-name" // Nombre del bucket en S3

val s3KeyPrefix = "path/to/upload" // Prefijo de ruta en S3 para los archivos

uploadCsvFiles(localDirectoryPath, bucketName, s3KeyPrefix)

}

Generar un fichero CSV en S3 como un flujo

noreply@blogger.com (fglopez) — Thu, 08 Jun 2023 10:42:00 +0000

Para crear un cliente S3:

import software.amazon.awssdk.auth.credentials.{AwsCredentialsProvider, ProfileCredentialsProvider}

import software.amazon.awssdk.regions.Region

import software.amazon.awssdk.services.s3.{S3Client, S3ClientBuilder}

// Crear un proveedor de credenciales basado en el perfil

val credentialsProvider: AwsCredentialsProvider = ProfileCredentialsProvider.builder()

.profileName("dev-usa")

.build()

// Crear un cliente de Amazon S3 utilizando el builder

val s3ClientBuilder: S3ClientBuilder = S3Client.builder()

.region(Region.US_EAST_1)

.credentialsProvider(credentialsProvider)

// Construir el cliente de Amazon S3

val s3Client: S3Client = s3ClientBuilder.build()

// Usar el cliente de Amazon S3 para realizar operaciones en S3

// ...

Implmentar un generador CSV:

import java.io.PrintWriter

import software.amazon.awssdk.services.s3.{S3Client, S3Configuration}

import software.amazon.awssdk.services.s3.model.{PutObjectRequest, PutObjectResponse}

object CsvGenerator {

def main(args: Array[String]): Unit = {

val bucketName = "my-s3-bucket"

val key = "output.csv" // Ruta del fichero de salida en S3

val s3Client = S3Client.builder()

.region(Region.US_EAST_1) // Configura la región de tu bucket de S3

.serviceConfiguration(S3Configuration.builder()

.pathStyleAccessEnabled(true)

.build())

.build()

val writer = new PrintWriter(new S3OutputStream(s3Client, bucketName, key))

try {

// Escribir los datos en el fichero CSV

writer.println("Columna1,Columna2,Columna3")

writer.println("Valor1,Valor2,Valor3")

writer.println("Valor4,Valor5,Valor6")

// ...

println("Fichero CSV generado correctamente en S3.")

} catch {

case e: Exception => e.printStackTrace()

} finally {

writer.close()

}

class S3OutputStream(s3Client: S3Client, bucketName: String, key: String) extends java.io.OutputStream {

private val putObjectRequest = PutObjectRequest.builder()

.bucket(bucketName)

.key(key)

.build()

private val outputStream = s3Client.putObject(putObjectRequest)

override def write(b: Int): Unit = {

outputStream.write(b)

}

override def flush(): Unit = {

outputStream.flush()

}

override def close(): Unit = {

outputStream.close()

}

Exportación de metadata de datahub a S3

noreply@blogger.com (fglopez) — Sun, 04 Jun 2023 12:05:00 +0000

Supuesto práctico

Tengo una aplicación Scala que llama a una API externa (Datahub), procesa la respuesta JSON y genera un conjunto de ficheros en local. Se puede ejecutar esta aplicación en un cluster EMR Serverless?

Ejecución en EMR Serverless

No es posible ejecutar directamente una aplicación Scala que procesa datos en un cluster EMR Serverless. EMR Serverless es un servicio de AWS que se utiliza para ejecutar trabajos de procesamiento de datos basados en Apache Spark en forma de consultas SQL o scripts de Spark, pero no es adecuado para ejecutar aplicaciones de Scala independientes.

En EMR Serverless, se ejecutan consultas o trabajos Spark en un entorno administrado por AWS, sin necesidad de aprovisionar y administrar un clúster de EMR tradicional. Esto permite un procesamiento rápido y escalable de datos sin preocuparse por la infraestructura subyacente.

Si tienes una aplicación Scala que procesa datos y genera archivos localmente, puedes ejecutarla en una instancia de EC2 en lugar de utilizar EMR Serverless. En EC2, tienes más control sobre el entorno de ejecución y puedes instalar las dependencias necesarias, configurar los recursos de hardware y ejecutar tu aplicación Scala directamente.

Puedes utilizar una instancia de EC2 para ejecutar tu aplicación Scala y, si es necesario, interactuar con otros servicios de AWS como S3 para el almacenamiento de archivos o incluso utilizar servicios de procesamiento de datos como EMR si tu aplicación requiere un procesamiento más complejo.

En resumen, EMR Serverless está diseñado específicamente para trabajar con consultas y trabajos Spark, no para ejecutar aplicaciones Scala independientes. Para ejecutar una aplicación Scala que genera archivos localmente, te recomendaría utilizar una instancia de EC2 en AWS.

Ejecutar en una instancia EC2

Si deseas ejecutar tu proceso batch en una instancia EC2 y planificar su ejecución, puedes seguir los siguientes pasos:

Crear una instancia EC2: Provisiona una instancia EC2 en AWS con el sistema operativo de tu elección. Asegúrate de seleccionar una instancia con los recursos adecuados para ejecutar tu proceso, como capacidad de CPU, memoria y almacenamiento suficientes.
Configurar la instancia EC2: Accede a la instancia EC2 y configúrala según tus necesidades. Esto puede incluir la instalación de las dependencias necesarias para tu proceso batch, como el lenguaje de programación, las bibliotecas y cualquier otra herramienta requerida.
Desarrollar y configurar el proceso batch: Implementa tu proceso batch en la instancia EC2. Esto puede implicar la escritura de scripts o programas que llamen a la API de Datahub, procesen los datos y generen los archivos CSV. Asegúrate de que el proceso esté configurado correctamente y funcione según lo esperado.
Programar la ejecución del proceso: Utiliza una herramienta de planificación de tareas en la instancia EC2, como cron en sistemas basados en Unix o el Programador de tareas en Windows, para programar la ejecución periódica de tu proceso. Configura la programación según tus necesidades, como la frecuencia y el horario de ejecución.
Configurar permisos y acceso a recursos: Asegúrate de que la instancia EC2 tenga los permisos adecuados para acceder a la API de Datahub y para escribir los archivos CSV en el almacenamiento deseado. Esto puede implicar la configuración de roles de IAM y asignarlos a la instancia EC2 para otorgar los permisos necesarios.
Monitorear y administrar la instancia EC2: Mantén un monitoreo regular de la instancia EC2 para asegurarte de que está funcionando correctamente. Puedes utilizar servicios como Amazon CloudWatch para recopilar métricas y registros de la instancia y configurar alarmas en caso de cualquier problema.

Con estos pasos, podrás ejecutar tu proceso batch en una instancia EC2 y planificar su ejecución según tus necesidades. Ten en cuenta que en este enfoque, la responsabilidad de la administración y escalado de la instancia EC2 recae en ti, a diferencia de los servicios administrados como AWS Lambda.

Ejecutar mediante función Lambda

Para ejecutar tu proceso batch que llama a una API de Datahub, procesa los datos y genera archivos CSV en AWS, puedes considerar los siguientes pasos:

Crear una función Lambda: Crea una función Lambda en AWS que contenga la lógica de tu proceso batch. Esta función será responsable de llamar a la API de Datahub, procesar los datos y generar los archivos CSV. Puedes desarrollar tu función en el lenguaje de programación de tu elección (por ejemplo, Python, Node.js, Java, etc.) y utilizar las bibliotecas necesarias para realizar las llamadas a la API y procesar los datos.
Configurar un trigger de tiempo: Utiliza el servicio CloudWatch Events para programar la ejecución periódica de tu función Lambda. Puedes configurar una regla en CloudWatch Events para que se active en un intervalo específico, por ejemplo, cada día a una hora determinada. Cuando se active la regla, CloudWatch Events invocará automáticamente tu función Lambda.
Configurar permisos y roles: Asegúrate de que la función Lambda tenga los permisos adecuados para acceder a la API de Datahub y para escribir los archivos CSV en el almacenamiento deseado, como Amazon S3 o Amazon EFS. Puedes crear un rol de IAM con las políticas necesarias y asignarlo a tu función Lambda.
Almacenar archivos CSV: Decide dónde quieres almacenar los archivos CSV generados por tu proceso batch. Puedes utilizar Amazon S3 para almacenarlos de forma duradera y segura. Configura tu función Lambda para guardar los archivos CSV en el bucket de Amazon S3 correspondiente.

Con estos pasos, tu proceso batch se ejecutará automáticamente cada día dentro de AWS. La función Lambda se activará según la programación establecida en CloudWatch Events, realizará las llamadas a la API de Datahub, procesará los datos y guardará los archivos CSV en Amazon S3. Esto te proporcionará una solución escalable, administrada y automatizada para ejecutar tu proceso batch en la nube de AWS.

Migración a Java 11

noreply@blogger.com (fglopez) — Sat, 27 May 2023 14:55:00 +0000

Para migrar tu componente Scala que utiliza la librería Spark de JDK 8 a JDK 11, hay algunas modificaciones que puedes considerar. A continuación, se detallan los cambios más relevantes:

Actualiza la versión de Scala: Asegúrate de utilizar una versión de Scala compatible con JDK 11. Puedes verificar la compatibilidad consultando la documentación de Scala. En tu archivo pom.xml o en la configuración del proyecto, actualiza la propiedad scala.version con la versión correspondiente.
Actualiza la versión de Spark: Verifica que estás utilizando una versión de Spark compatible con JDK 11. En tu archivo pom.xml o en la configuración del proyecto, actualiza las dependencias de Spark (por ejemplo, spark-core, spark-sql, spark-avro) con la versión adecuada.
Actualiza la configuración de la JDK: En IntelliJ, ve a "File" -> "Project Structure" -> "Project Settings" -> "Project" y selecciona la JDK 11 como el SDK del proyecto.
Actualiza las dependencias: Verifica que todas las dependencias utilizadas en tu proyecto sean compatibles con JDK 11. Actualiza las versiones si es necesario y asegúrate de utilizar bibliotecas actualizadas y compatibles con la nueva versión de Java.
Revisa cambios en la API de Spark: Es posible que algunas clases, métodos o configuraciones hayan cambiado en las versiones más recientes de Spark. Verifica la documentación oficial de Spark para identificar posibles cambios y ajusta tu código en consecuencia.
Verifica el código relacionado con la gestión de recursos: En JDK 11, hay algunos cambios en la gestión de recursos, como el uso del bloque try-with-resources para asegurarte de que los recursos sean cerrados correctamente. Verifica tu código y aplica los cambios necesarios en la gestión de recursos.
Considera migrar a Java 11 en lugar de Scala: Si tu código Scala no hace un uso intensivo de características específicas de Scala y la migración a JDK 11 implica demasiados cambios, también puedes considerar migrar a Java 11. Esto puede simplificar la migración y garantizar la compatibilidad con JDK 11 de manera más directa.

Recuerda que es importante realizar pruebas exhaustivas después de la migración para asegurarte de que tu componente Scala funcione correctamente con la nueva JDK 11.

Recursos adicionales

https://www.adictosaltrabajo.com/2019/05/21/migracion-de-java-8-a-java-11/

https://docs.scala-lang.org/overviews/jdk-compatibility/overview.html#inner-main

https://medium.com/globant/migrating-to-java-11-from-java-8-bd6343107ea6

Observatorio BI & Analytics

La deshumanización de la IA

Guía Desarrollo Spark

Tabla de Monitorización de tareas semanales

Septiembre 2023

Octubre 2023

Noviembre 2023

Plantilla de Peticiones JIRA para Data Engineering

Impresión de esquemas complejos de un Dataframe Spark

POO en Scala: traits, mixins y herencia

CLASSES AND OBJECTS

TRAITS

MIXINS

Inheritance

CONCLUSION

Crear bucle for/yield en Scala

Problem

Solution

Discussion

See Also

Pattern matching en Scala

Programming in Scala, Third Edition Resources

(If you don't already have it, you can purchase Programming in Scala, Third Edition here.)

Pattern Matching | Scala Documentation

Pattern matching is a mechanism for checking a value against a pattern. A successful match can also deconstruct a value…

Formatear código en IntelliJ

Recursos Data Engineer

Transformar dataframe spark

Extraer partes del nombre de una entidad MySQL

Comparación de esquema entre Parquet y Athena

Prueba de lectura de Parameter Store en ScalaTest

Pruebas unitarias de llamada a APIs en ScalaTest

Pruebas unitarias de tipo mock en ScalaTest (SsmUtils)

Generar un ZIP de objetos S3 con un prefijo de clave comun

Procesamiento de peticiones GraphQL en Datahub

Instanciacion de cliente S3 con SDK de AWS

Generador de ficheros en S3

Obtener todos los ficheros CSV dentro de un directorio

Subida recursiva de archivos CSV en un directorio local a S3

Generar un fichero CSV en S3 como un flujo

Exportación de metadata de datahub a S3

Migración a Java 11