Primeros pasos con Apache Spark 2

Posted on August 10, 2016 by admin

Hace pocos días salió la esperada versión 2 de Apache Spark y como algunos de ustedes saben es un framework que ahora mismo atrae mucho mi atención y como no pudo ser de otra forma hice un pequeño proyecto donde quiero ir colocando ejemplos sencillos de Spark con las nuevas (y no tan nuevas) cosas de Spark.

Para empezar comentarles que yo todavía no he utilizado sbt sino por el contrario uso maven como herramienta de construcción de proyectos. He aquí los primeros cambios necesarios para trabajar con spark 2, las dependencias correspondientes a la versión (indicadas en el pom.xml).

   <dependencies>
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-core_2.10</artifactId>
      <version>2.0.0</version>
    </dependency>

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-sql_2.10</artifactId>
      <version>2.0.0</version>
    </dependency>

   ...

<groupId>org.apache.spark</groupId>

<artifactId>spark-core_2.10</artifactId>

</dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-sql_2.10</artifactId>

</dependency>

...

Entre los nuevos cambios de spark está que el punto de entrada para los programas spark ya no serán el hiveContext o sqlContext sino que han sido subsumidas en una clase llamada SparkSession. Las clases HiveContext y SQLContext se han mantenido para proporcionar retrocompatibilidad. Ejemplo

// En spark v1.6.2 &lt;= lo hubieramos hecho asi
  //  val conf = new SparkConf().setMaster("local").setAppName("Simple Application")
  //  val sc = new SparkContext(conf)
  //  val sqlContext = new org.apache.spark.sql.SQLContext(sc)
  val sparkSession = SparkSession.builder.
    master("local")
    .appName("Simple Application")
    .getOrCreate()

// En spark v1.6.2 <= lo hubieramos hecho asi

// val conf = new SparkConf().setMaster("local").setAppName("Simple Application")

// val sc = new SparkContext(conf)

// val sqlContext = new org.apache.spark.sql.SQLContext(sc)

val sparkSession = SparkSession.builder.

master("local")

.appName("Simple Application")

.getOrCreate()

Con el SparkSession haremos lo mismo que hacíamos con sqlContext por ejemplo obtener un Dataset

  //obtenemos el dataset de tipo People
  val ds = sparkSession.read.json("src/main/resources/people.json").as[People]

1 2	//obtenemos el dataset de tipo People val ds = sparkSession.read.json("src/main/resources/people.json").as[People]

O por el contrario obtener un DataFrame

val df = sparkSession.read.format("com.databricks.spark.csv")
    .format("com.databricks.spark.csv")
    .option("header", "true") // Use first line of all files as header
    .option("inferSchema", "true") // Automatically infer data types
    .load("src/main/resources/datos.csv")

val df = sparkSession.read.format("com.databricks.spark.csv")

.format("com.databricks.spark.csv")

.option("header", "true") // Use first line of all files as header

.option("inferSchema", "true") // Automatically infer data types

.load("src/main/resources/datos.csv")

Otro punto importante ha sido la unificación de las clases Dataset y DataFrame (para Java y Scala) a partir de la versión 2.¿Qué significa esto? pues sencillamente que ahora solo existirá la clase Dataset, pero proporcionará la misma funcionalidad que nos daba la clase DataFrame, de hecho basta con comparar la API en la versión 1.6.2 y 2.0.0 y ver como los métodos de la clase DataFrame están ahora incluidos en la clase Dataset.

Dataset y Dataframe en Spark 2

Aquellos interesados en leer más acerca de Dataset y Dataframe visitar este link

Estos no son los únicos cambios en Spark, de hechos son muchos más, que se corresponden a optimizaciones a nivel de compilación y ejecución así como también a un nuevo parseador SQL, para leer mas acerca de lo nuevo en Spark 2 clic aquí.

Aqui les dejo en enlace al proyecto donde ire añadiendo clases y seguiré probando mas cosas nuevas de Spark.

¿Valen la pena las certificaciones?

Posted on July 18, 2016 by admin

Tiempo atrás me hicieron la siguiente consulta en el blog:

“Hola Jose,

estoy muy indecisa en si prepararme para la certificación de Java. Mucha gente me dice que es absurdo porque las preguntas son del api, y tipo trozos de código que tienes que hacer de compilador y que al final se vuelve en un absurdo porque eso al fin y al cabo en nuestro día a día con un compilador es intuitivo. También me han dicho que a la hora del mundo laboral apenas lo valoran porque como debemos pagar para certificarnos, como que los empresarios lo ven como un título comprado.“

Antes que nada disculparme con la persona que me escribió ya que ha pasado algo de tiempo y apenas ahora es que le respondo, pero aquí voy y con la sinceridad que me caracteriza.

En más de una vez he recalcado estar a favor en realizar certificaciones, pero así como lo he dicho alguna vez, ahora lo repito, “no es por el papel”, el papel o el certificado es lo de menos, bueno (pausa de 1 seg), es un premio o mejor dicho un reconocimiento al esfuerzo y a la dedicación que has puesto en estudiar y aprender cierta tecnología o idioma, es como cuando presentábamos un examen importante (para obtener créditos especiales o en una materia electiva de particular importancia) y después de tanto estudiar te otorgaban un aprobado pero que goza de cierto cache, ya que no se aprueba con la mitad de la nota sino con un poco más (por ejemplo si las notas van de 1 a 10 se otorgue únicamente el aprobado con 6.5). Lo que a mí particularmente me parece fascinante de las certificaciones y me entusiasma a realizarlas es todo el proceso de preparación, es como te planteas y organizas los objetivos para estudiarlos, como al cabo de un tiempo empiezas a darte cuenta de que algunos temas los tienes bien cubiertos mientras que hay otros donde necesitas hacer énfasis y repasarlos de nuevo, todo ese proceso de estudio con ese poquitín de presión añadida pero sobretodo con la ambición de aprender y mejorar tus habilidades profesionales, creo que hace méritos más que suficientes para presentar una certificación, y voy a plantear par de ejemplos para intentar hacerles llegar mi punto de vista.

Imaginemonos que empezamos a prepararnos durante algunos meses en presentar un examen de inglés, por ejemplo el FCE (también llamado el first). Si nos hemos esforzado digamos que durante unos 5 meses estudiando, leyendo muchos verbos, conjungandolos y utilizándolos en frases, escuchando mucho inglés, etc… ya habremos ganado, porque independientemente de que al presentar aprobemos o no el examen, el premio lo recibiremos cuando nos toque hablar con una persona de habla inglesa y hayamos podido entender a esta persona y a su vez hayamos logrado expresarnos con cierta fluidez, que si no fue tal, pues sencillamente no importa porque estoy seguro de que nos expresaríamos mucho mejor que antes de haber empezado a estudiar para la certificación.

Otra razón de presentar las certificaciones, es el saber, porque para mí “el saber es poder”, el saber te da confianza, y si vas a presentarte para una entrevista de trabajo o para un cargo, créeme que la confianza es importante (te desenvuelves mejor, yo sinceramente creo que hasta te cambia la cara y hasta la forma de expresarte) y ahora bien ¿recuerdas cuando en la escuela tenías que hacer una exposición? a la mayoría nos daba algo de miedo escénico, pero había una cosa que siempre era palpable independientemente de que fueras el mejor en el escenario o no, y es que siempre se sabía quien había estudiado y quien no. ¿Crees que las personas de RRHH no se dan cuenta? y si tuviste suerte y pasaste esa entrevista recuerda que luego seguramente tendrás una entrevista técnica y a esta persona será difícil que le engañes o mejor dicho dejaras mejor imagen si le demuestras a esa persona que sabes de lo que hablas.

Entonces, ¿valen la pena las certificaciones? SI

¿Los empresarios lo ven como un titulo comprado? No conozco ninguno que lo vea de esa manera y en dado caso desde mi humilde punto de vista te diría corre y vete de esa empresa porque allí no te valorarán ni como profesional ni tu deseo de ser mejor.

Yo particularmente he visto como últimamente empresas dedican una “bolsa” de dinero para fomentar la certificación de sus empleados, ya que a la hora de escoger un perfil entre un grupo de candidatos, la certificación y el haber aprobado la entrevista técnica puede que pongan la balanza a tu favor. Además no olvidemos que las empresas son las principales beneficiadas de que nosotros los empleados nos certifiquemos, ya que a priori están ganando, teniendo en el staff mejores profesionales.

¿Algunas preguntas son muy complejas y debemos actuar como si fuésemos un compilador, ya que incluyen código? Si hay preguntas que incluyen código y debemos actuar como un compilador pero ni son muchas y tampoco todas son extremadamente difíciles, estas la mayoría de las veces son conchas de mango, que si has estudiado, las vas a pillar de inmediato.

Así que amiga no lo pienses 2 veces certificate si tienes interés en hacerlo, que en el peor de los casos lo que harás es aprender 😉 y serás bien valorada por el gremio profesional y por tus colegas por interesante en afrontar retos de este tipo.

Por último confesar que ya he empezado a preparar mi próxima certificación y como adelanto les dire que va enfocada en tecnologías Big Data.

PD: La de Java 8 me hace mucho ojito, jejejeje, pero ya he estudiado por mi parte las expresiones lambda, los streams, los default methods de las interfaces, entre otras cosas, por lo cual no siento tanto interés como otras veces en presentarla, así que quizás cuando salga Java 9 me meta de cabeza a estudiar de nuevo.

Representación gráfica de mi cuenta de Twitter

Posted on July 15, 2016 by admin

Hola de nuevo chicos, este es un post breve pero que quería compartir con ustedes desde hace mucho tiempo, fue hace un año aproximadamente que en la materia de análisis de redes sociales realicé un mini proyecto, este consistió en analizar la actividad de mi red social de twitter, y además tener una representación gráfica de la actividad de mi cuenta, es decir, a quienes sigo y las menciones y hashtags utilizados por mí y por aquellos a quienes sigo, incluso el número de veces que se ha utilizado cada hashtag.

Lo único que he hecho antes de subir el código a github fue actualizar la versión neo4j a 2.3.6 (base de datos de grafo donde se guardan las relaciones) y eliminar mis datos para el uso de la API de twitter.

Antes de ejecutar esta aplicación será necesario que generen un token y sigan los pasos necesarios para poder utilizar la API de twitter, por otro lado es importante destacar que este programa almacena las distintas relaciones entre entidades en una base de datos embebida de neo4j y para poder visualizar el resultado final de todas esas relaciones guardadas, lo que hice fue simplemente utilizar el navegador/visualizador por defecto que trae neo4j (que si no me equivoco esta creado con d3.js). Así que manos a la obra y comencemos descargando e instalando la versión 2.3.6 de Neo4j desde el siguiente enlace y una vez realizada la instalación pasamos al código fuente.

A continuación la clase principal:

package com.josedeveloper.twitter;

import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import javax.json.Json;
import javax.json.JsonArray;
import javax.json.JsonObject;
import javax.json.JsonReader;
import javax.json.JsonValue;

import oauth.signpost.OAuthConsumer;
import oauth.signpost.commonshttp.CommonsHttpOAuthConsumer;
import oauth.signpost.exception.OAuthCommunicationException;
import oauth.signpost.exception.OAuthExpectationFailedException;
import oauth.signpost.exception.OAuthMessageSignerException;

import org.apache.http.HttpResponse;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.HttpClientBuilder;
import org.neo4j.graphdb.GraphDatabaseService;
import org.neo4j.graphdb.Label;
import org.neo4j.graphdb.Node;
import org.neo4j.graphdb.Relationship;
import org.neo4j.graphdb.RelationshipType;
import org.neo4j.graphdb.Transaction;
import org.neo4j.graphdb.factory.GraphDatabaseFactory;

public class TwitterRelationshipApp {

	static final String TWITTER_DB_PATH = "DATABASE_PATH";
	static final String CONSUMER_KEY = "YOUR_CONSUMER_KEY";
	static final String CONSUMER_SECRET = "YOUR_CONSUMER_SECRET";
	static final String ACCESS_TOKEN = "YOUR_ACCESS_TOKEN";
	static final String ACCESS_TOKEN_SECRET = "YOUR_ACCESS_TOKEN_SECRET";

	private final GraphDatabaseService graphDB;
	private final Set totalUsers;
	private final String account;
	private final int count;

	enum NodeType implements Label {
		TWITTER_USER, HASHTAG;
	}

	enum Relationships implements RelationshipType {
		USE, MENTION;
	}

	public TwitterRelationshipApp(final String account, final int count) {
		this.account = account;
		this.count = count;

		totalUsers = new HashSet&lt;&gt;();
		graphDB = new GraphDatabaseFactory().newEmbeddedDatabase(new File(TWITTER_DB_PATH));
	}

    public static void main(String[] args) throws OAuthMessageSignerException, OAuthExpectationFailedException, OAuthCommunicationException, ClientProtocolException, IOException
    {
    	TwitterRelationshipApp app = new TwitterRelationshipApp("josedeveloper", 100);
    	app.registerShutdownHook();
    	app.insertUsers();
    	app.insertUserMentionsRelationshipsByUser();
    }

	private void insertUsers() throws OAuthMessageSignerException, OAuthExpectationFailedException, OAuthCommunicationException, ClientProtocolException, IOException {
    	OAuthConsumer oAuthConsumer = new CommonsHttpOAuthConsumer(CONSUMER_KEY, CONSUMER_SECRET);
		oAuthConsumer.setTokenWithSecret(ACCESS_TOKEN, ACCESS_TOKEN_SECRET);

		HttpGet httpGet = new HttpGet("https://api.twitter.com/1.1/friends/list.json?screen_name=" + account + "&amp;count=" + count); //those who I follow

		oAuthConsumer.sign(httpGet);

		HttpClient httpClient = HttpClientBuilder.create().build();
		HttpResponse httpResponse = httpClient.execute(httpGet);

		//int statusCode = httpResponse.getStatusLine().getStatusCode();

		JsonReader reader = Json.createReader(httpResponse.getEntity().getContent());
		JsonObject root = reader.readObject();
		JsonArray users = root.getJsonArray("users");

		Iterator iter = users.iterator();
		while (iter.hasNext()) {
			JsonObject user = (JsonObject) iter.next();

			try (Transaction tx = graphDB.beginTx()) {
				Node userNode = graphDB.createNode(NodeType.TWITTER_USER);

				userNode.setProperty("id", user.getString("id_str"));
				userNode.setProperty("name", user.getString("name"));
				userNode.setProperty("screen_name", user.getString("screen_name"));

				insertRelationshipsWithHashtagsByUser(userNode, graphDB);

				tx.success();
			} catch (Exception e) {
				System.out.println(e);
			}

			totalUsers.add(user.getString("screen_name"));
		}
	}

	private void registerShutdownHook() {


		    // Registers a shutdown hook for the Neo4j instance so that it
		    // shuts down nicely when the VM exits (even if you "Ctrl-C" the
		    // running application).
		Runtime.getRuntime().addShutdownHook(new Thread() {
			@Override
		    public void run() {
				graphDB.shutdown();
		    }

		});
	}


	private static void insertRelationshipsWithHashtagsByUser(Node user, final GraphDatabaseService db) throws OAuthMessageSignerException, OAuthExpectationFailedException, OAuthCommunicationException, ClientProtocolException, IOException {
		OAuthConsumer oAuthConsumer = new CommonsHttpOAuthConsumer(CONSUMER_KEY, CONSUMER_SECRET);
		oAuthConsumer.setTokenWithSecret(ACCESS_TOKEN, ACCESS_TOKEN_SECRET);

		HttpGet httpGet = new HttpGet("https://api.twitter.com/1.1/statuses/user_timeline.json?screen_name=" + user.getProperty("screen_name"));
		oAuthConsumer.sign(httpGet);

		HttpClient httpClient = HttpClientBuilder.create().build();
		HttpResponse httpResponse = httpClient.execute(httpGet);

		//int statusCode = httpResponse.getStatusLine().getStatusCode();

		JsonReader timelineReader = Json.createReader(httpResponse.getEntity().getContent());
		JsonArray tweets = timelineReader.readArray();
		Iterator tweetsIter = tweets.iterator();
		Map&lt;String, Integer&gt; usedHashtags = new HashMap&lt;&gt;();
		while(tweetsIter.hasNext()) {
			JsonObject tweet = (JsonObject) tweetsIter.next();

			JsonObject entities = tweet.getJsonObject("entities");
			JsonArray hashtags = entities.getJsonArray("hashtags");
			Iterator hashtagsIter = hashtags.iterator();

			while (hashtagsIter.hasNext()) {
				String hashtag = ((JsonObject) hashtagsIter.next()).getString("text");

				if (usedHashtags.containsKey(hashtag)) {
					Integer counter = usedHashtags.get(hashtag);
					usedHashtags.put(hashtag, ++counter);
				} else{
					usedHashtags.put(hashtag, Integer.valueOf(1));
				}
			}

		}

		for (String hashtag : usedHashtags.keySet()) {

			try (Transaction tx = db.beginTx()) {
				Node hashtagNode = db.findNode(NodeType.HASHTAG, "text", hashtag);
				if (hashtagNode == null)
					hashtagNode = db.createNode(NodeType.HASHTAG);

				hashtagNode.setProperty("text", hashtag);

				Integer timesUsed = usedHashtags.get(hashtag);
				Relationship use = user.createRelationshipTo(hashtagNode, Relationships.USE);
				use.setProperty("times", timesUsed);

				tx.success();
			} catch (Exception e) {
				System.out.println(e);
			}
		}


	}

	private void insertUserMentionsRelationshipsByUser() throws OAuthMessageSignerException, OAuthExpectationFailedException, OAuthCommunicationException, ClientProtocolException, IOException {
		OAuthConsumer oAuthConsumer = new CommonsHttpOAuthConsumer(CONSUMER_KEY, CONSUMER_SECRET);
		oAuthConsumer.setTokenWithSecret(ACCESS_TOKEN, ACCESS_TOKEN_SECRET);

		for (String twitterUser : totalUsers) {

			HttpGet httpGet = new HttpGet("https://api.twitter.com/1.1/statuses/user_timeline.json?screen_name=" + twitterUser);
			oAuthConsumer.sign(httpGet);

			HttpClient httpClient = HttpClientBuilder.create().build();
			HttpResponse httpResponse = httpClient.execute(httpGet);

			//int statusCode = httpResponse.getStatusLine().getStatusCode();

			JsonReader timelineReader = Json.createReader(httpResponse.getEntity().getContent());
			JsonArray tweets = timelineReader.readArray();
			Iterator tweetsIter = tweets.iterator();
			Map&lt;String, Integer&gt; userMentionsDone = new HashMap&lt;&gt;();
			while(tweetsIter.hasNext()) {
				JsonObject tweet = (JsonObject) tweetsIter.next();

				JsonObject entities = tweet.getJsonObject("entities");
				JsonArray userMentions = entities.getJsonArray("user_mentions");
				Iterator hashtagsIter = userMentions.iterator();

				while (hashtagsIter.hasNext()) {
					String userMentioned = ((JsonObject) hashtagsIter.next()).getString("screen_name");

					if (totalUsers.contains(userMentioned)) {
						if (userMentionsDone.containsKey(userMentioned)) {
							Integer counter = userMentionsDone.get(userMentioned);
							userMentionsDone.put(userMentioned, ++counter);
						} else{
							userMentionsDone.put(userMentioned, Integer.valueOf(1));
						}
					}
				}

			}

			for (String userMentionDone : userMentionsDone.keySet()) {

				try (Transaction tx = graphDB.beginTx()) {
					Node twitterUserMentionedNode = graphDB.findNode(NodeType.TWITTER_USER, "screen_name", userMentionDone);
					Node twitterUserNode = graphDB.findNode(NodeType.TWITTER_USER, "screen_name", twitterUser);

					Integer timesMentioned = userMentionsDone.get(userMentionDone);
					Relationship use = twitterUserNode.createRelationshipTo(twitterUserMentionedNode, Relationships.MENTION);
					use.setProperty("times", timesMentioned);

					tx.success();
				} catch (Exception e) {
					System.out.println(e);
				}
			}
		}

	}

}

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

package com.josedeveloper.twitter;

import java.io.File;

import java.io.IOException;

import java.util.HashMap;

import java.util.HashSet;

import java.util.Iterator;

import java.util.Map;

import java.util.Set;

import javax.json.Json;

import javax.json.JsonArray;

import javax.json.JsonObject;

import javax.json.JsonReader;

import javax.json.JsonValue;

import oauth.signpost.OAuthConsumer;

import oauth.signpost.commonshttp.CommonsHttpOAuthConsumer;

import oauth.signpost.exception.OAuthCommunicationException;

import oauth.signpost.exception.OAuthExpectationFailedException;

import oauth.signpost.exception.OAuthMessageSignerException;

import org.apache.http.HttpResponse;

import org.apache.http.client.ClientProtocolException;

import org.apache.http.client.HttpClient;

import org.apache.http.client.methods.HttpGet;

import org.apache.http.impl.client.HttpClientBuilder;

import org.neo4j.graphdb.GraphDatabaseService;

import org.neo4j.graphdb.Label;

import org.neo4j.graphdb.Node;

import org.neo4j.graphdb.Relationship;

import org.neo4j.graphdb.RelationshipType;

import org.neo4j.graphdb.Transaction;

import org.neo4j.graphdb.factory.GraphDatabaseFactory;

public class TwitterRelationshipApp {

static final String TWITTER_DB_PATH = "DATABASE_PATH";

static final String CONSUMER_KEY = "YOUR_CONSUMER_KEY";

static final String CONSUMER_SECRET = "YOUR_CONSUMER_SECRET";

static final String ACCESS_TOKEN = "YOUR_ACCESS_TOKEN";

static final String ACCESS_TOKEN_SECRET = "YOUR_ACCESS_TOKEN_SECRET";

private final GraphDatabaseService graphDB;

private final Set totalUsers;

private final String account;

private final int count;

enum NodeType implements Label {

TWITTER_USER, HASHTAG;

}

enum Relationships implements RelationshipType {

USE, MENTION;

}

public TwitterRelationshipApp(final String account, final int count) {

this.account = account;

this.count = count;

totalUsers = new HashSet<>();

graphDB = new GraphDatabaseFactory().newEmbeddedDatabase(new File(TWITTER_DB_PATH));

}

public static void main(String[] args) throws OAuthMessageSignerException, OAuthExpectationFailedException, OAuthCommunicationException, ClientProtocolException, IOException

{

TwitterRelationshipApp app = new TwitterRelationshipApp("josedeveloper", 100);

app.registerShutdownHook();

app.insertUsers();

app.insertUserMentionsRelationshipsByUser();

}

private void insertUsers() throws OAuthMessageSignerException, OAuthExpectationFailedException, OAuthCommunicationException, ClientProtocolException, IOException {

OAuthConsumer oAuthConsumer = new CommonsHttpOAuthConsumer(CONSUMER_KEY, CONSUMER_SECRET);

oAuthConsumer.setTokenWithSecret(ACCESS_TOKEN, ACCESS_TOKEN_SECRET);

HttpGet httpGet = new HttpGet("https://api.twitter.com/1.1/friends/list.json?screen_name=" + account + "&count=" + count); //those who I follow

oAuthConsumer.sign(httpGet);

HttpClient httpClient = HttpClientBuilder.create().build();

HttpResponse httpResponse = httpClient.execute(httpGet);

//int statusCode = httpResponse.getStatusLine().getStatusCode();

JsonReader reader = Json.createReader(httpResponse.getEntity().getContent());

JsonObject root = reader.readObject();

JsonArray users = root.getJsonArray("users");

Iterator iter = users.iterator();

while (iter.hasNext()) {

JsonObject user = (JsonObject) iter.next();

try (Transaction tx = graphDB.beginTx()) {

Node userNode = graphDB.createNode(NodeType.TWITTER_USER);

userNode.setProperty("id", user.getString("id_str"));

userNode.setProperty("name", user.getString("name"));

userNode.setProperty("screen_name", user.getString("screen_name"));

insertRelationshipsWithHashtagsByUser(userNode, graphDB);

tx.success();

} catch (Exception e) {

System.out.println(e);

}

totalUsers.add(user.getString("screen_name"));

}

private void registerShutdownHook() {

// Registers a shutdown hook for the Neo4j instance so that it

// shuts down nicely when the VM exits (even if you "Ctrl-C" the

// running application).

Runtime.getRuntime().addShutdownHook(new Thread() {

@Override

public void run() {

graphDB.shutdown();

}

});

}

private static void insertRelationshipsWithHashtagsByUser(Node user, final GraphDatabaseService db) throws OAuthMessageSignerException, OAuthExpectationFailedException, OAuthCommunicationException, ClientProtocolException, IOException {

OAuthConsumer oAuthConsumer = new CommonsHttpOAuthConsumer(CONSUMER_KEY, CONSUMER_SECRET);

oAuthConsumer.setTokenWithSecret(ACCESS_TOKEN, ACCESS_TOKEN_SECRET);

HttpGet httpGet = new HttpGet("https://api.twitter.com/1.1/statuses/user_timeline.json?screen_name=" + user.getProperty("screen_name"));

oAuthConsumer.sign(httpGet);

HttpClient httpClient = HttpClientBuilder.create().build();

HttpResponse httpResponse = httpClient.execute(httpGet);

//int statusCode = httpResponse.getStatusLine().getStatusCode();

JsonReader timelineReader = Json.createReader(httpResponse.getEntity().getContent());

JsonArray tweets = timelineReader.readArray();

Iterator tweetsIter = tweets.iterator();

Map<String, Integer> usedHashtags = new HashMap<>();

while(tweetsIter.hasNext()) {

JsonObject tweet = (JsonObject) tweetsIter.next();

JsonObject entities = tweet.getJsonObject("entities");

JsonArray hashtags = entities.getJsonArray("hashtags");

Iterator hashtagsIter = hashtags.iterator();

while (hashtagsIter.hasNext()) {

String hashtag = ((JsonObject) hashtagsIter.next()).getString("text");

if (usedHashtags.containsKey(hashtag)) {

Integer counter = usedHashtags.get(hashtag);

usedHashtags.put(hashtag, ++counter);

} else{

usedHashtags.put(hashtag, Integer.valueOf(1));

}

for (String hashtag : usedHashtags.keySet()) {

try (Transaction tx = db.beginTx()) {

Node hashtagNode = db.findNode(NodeType.HASHTAG, "text", hashtag);

if (hashtagNode == null)

hashtagNode = db.createNode(NodeType.HASHTAG);

hashtagNode.setProperty("text", hashtag);

Integer timesUsed = usedHashtags.get(hashtag);

Relationship use = user.createRelationshipTo(hashtagNode, Relationships.USE);

use.setProperty("times", timesUsed);

tx.success();

} catch (Exception e) {

System.out.println(e);

}

private void insertUserMentionsRelationshipsByUser() throws OAuthMessageSignerException, OAuthExpectationFailedException, OAuthCommunicationException, ClientProtocolException, IOException {

OAuthConsumer oAuthConsumer = new CommonsHttpOAuthConsumer(CONSUMER_KEY, CONSUMER_SECRET);

oAuthConsumer.setTokenWithSecret(ACCESS_TOKEN, ACCESS_TOKEN_SECRET);

for (String twitterUser : totalUsers) {

HttpGet httpGet = new HttpGet("https://api.twitter.com/1.1/statuses/user_timeline.json?screen_name=" + twitterUser);

oAuthConsumer.sign(httpGet);

HttpClient httpClient = HttpClientBuilder.create().build();

HttpResponse httpResponse = httpClient.execute(httpGet);

//int statusCode = httpResponse.getStatusLine().getStatusCode();

JsonReader timelineReader = Json.createReader(httpResponse.getEntity().getContent());

JsonArray tweets = timelineReader.readArray();

Iterator tweetsIter = tweets.iterator();

Map<String, Integer> userMentionsDone = new HashMap<>();

while(tweetsIter.hasNext()) {

JsonObject tweet = (JsonObject) tweetsIter.next();

JsonObject entities = tweet.getJsonObject("entities");

JsonArray userMentions = entities.getJsonArray("user_mentions");

Iterator hashtagsIter = userMentions.iterator();

while (hashtagsIter.hasNext()) {

String userMentioned = ((JsonObject) hashtagsIter.next()).getString("screen_name");

if (totalUsers.contains(userMentioned)) {

if (userMentionsDone.containsKey(userMentioned)) {

Integer counter = userMentionsDone.get(userMentioned);

userMentionsDone.put(userMentioned, ++counter);

} else{

userMentionsDone.put(userMentioned, Integer.valueOf(1));

}

for (String userMentionDone : userMentionsDone.keySet()) {

try (Transaction tx = graphDB.beginTx()) {

Node twitterUserMentionedNode = graphDB.findNode(NodeType.TWITTER_USER, "screen_name", userMentionDone);

Node twitterUserNode = graphDB.findNode(NodeType.TWITTER_USER, "screen_name", twitterUser);

Integer timesMentioned = userMentionsDone.get(userMentionDone);

Relationship use = twitterUserNode.createRelationshipTo(twitterUserMentionedNode, Relationships.MENTION);

use.setProperty("times", timesMentioned);

tx.success();

} catch (Exception e) {

System.out.println(e);

}

El código completo del proyecto pueden hallarlo en el siguiente enlace. Una vez hayamos ejecutado nuestra aplicación, en la ruta especificada TWITTER_DB_PATH encontraremos una carpeta con extensión .db donde estarán almacenadas las relaciones (todo el grafo), el siguiente paso para poder visualizar el grafo será editar el fichero RUTA_INSTALACION_NEO4J/conf/neo4j-server.properties y editar la ruta donde ha de estar ubicada la base de datos

org.neo4j.server.database.location=TWITTER_DB_PATH.db

TWITTER_DB_PATH= La ruta especificada donde se ha de crear la base de datos Neo4j donde se almacenarán las relaciones.

Ahora procedemos a arrancar la base de datos, que es bastante sencillo solo es necesario ejecutar el siguiente comando:

RUTA_INSTALACION_NEO4J/bin/neo4j start

Hecho esto desde un navegador (chrome o firefox por ejemplo) ir a la ruta http://localhost:7474 y veremos el cliente web de Neo4j

neo4j

Luego desde la consola donde ejecutar las consultas (donde aparece el símbolo del $) ejecutar el siguiente comando para poder visualizar todo el grafo.

MATCH (n) RETURN n

En mi caso obtuve lo siguiente

relaciones neo4j

Como podrán darse cuenta se pueden distinguir los distintos tipos de relaciones (USE y MENTION) entre los distintos nodos, además hay 2 tipos de nodos, los azules son las cuentas de twitter y los verdes son los hashtags.

Otra cosa interesante es que con el visualizador de Neo4j podemos ver los datos de las relaciones como por ejemplo el número de veces que una cuenta de twitter ha usado un hashtag o mencionado a otra cuenta como en la siguiente gráfica

Número de veces que un hashtag es utilizado

Como se puede apreciar en la parte inferior de la gráfica, la cuenta Ben & Martijn ha utilizado 2 veces el hashtag #Java.

Bueno ya no me queda más nada que mostrar a este respecto, así que si te resulta interesante ejecuta este ejemplo y ve como es la actividad de tu cuenta de twitter y te aseguro que encontraras cosas que te llamarán la atención y si te parece compártelas con el resto.

Por último mencionarles que intentare en medida de lo posible actualizar este código para que trabaje con la versión 3 o superior de Neo4j, mejorar el código (hacerlo más claro) y actualizarlo a Java 8. Cualquier comentario y/o sugerencia soy todo oídos.

Primer ejemplo con apache Storm

Posted on July 10, 2016 by admin

Apache Storm es un framework de procesamiento distribuido de eventos. Empresas como twitter utilizaron Storm desde el 2011 aunque posteriormente lo reemplazó por Heron en el 2015.

Actualmente me encuentro trabajando en una aplicación construida con Apache Storm y quiero compartir con ustedes mi primer ejemplo con Apache Storm el cual es bastante simple pero cumplio su cometido que era el iniciarme en este framework y entender sus componentes principales.

Necesitaremos descargar rabbitMQ la versión 3.5.7.

Una vez hayamos instalado rabbitMQ (es solo cuestión de descomprimir) vamos a habilitar un plugin (un cliente web) donde podremos de forma sencilla monitorizar las colas, entonces nos ubicamos en la siguiente ruta ruta_instalacion_rabbitmq/sbin/ y desde allí ejecutamos el siguiente comando

./rabbitmq-plugins enable rabbitmq_management

Este comando habilitará el cliente web que nos permitirá monitorizar las colas, los exchanges incluso manipular las colas pudiendo ingresar elementos a las colas, sacar elementos e incluso purgar las colas. Paso siguiente iniciaremos el rabbitMQ con el siguiente comando:

./rabbitmq-server

Inmediatamente después desde un navegador nos dirigimos a la dirección http://localhost:15672

rabbitmq overview

Una vez allí crearemos una cola para hacer nuestro ejemplo y la llamaremos “data”.

agregar cola en rabbitmq

Ahora deberíamos ser capaces de ver la única cola de nuestro sistema. Clicando en ella podríamos incluso agregarle mensajes por medio de la interfaz gráfica (si lo desean hagan la prueba y verán como cada uno de los mensajes que agreguen se irán encolando), pero la inserción de mensajes en la cola lo haremos mediante un pequeño programa Java.

cola data

Ahora un poco de teoría para conocer acerca de Apache Storm.

¿Qué es Apache Storm?

Es un framework de computación distribuida en tiempo real, escrito en su mayoría en Clojure. Storm es similar a la forma como Hadoop ofrece un conjunto de primitivas generales para hacer el procesamiento por lotes, también ofrece un conjunto de primitivas generales para hacer cómputos en tiempo real. Storm es simple, se puede utilizar con cualquier lenguaje de programación.

Las aplicaciones de Storm son creadas como topologías en la forma de DAG (Directed Acyclic Graph) con spouts y bolts actuando como los vertices del grado. Las aristas en el grafo son llamados streams y dirigen la data de un nodo a otro. Juntos, la topología actúa como una tubería de transformación de datos.

Los spouts son fuentes de flujo (streams) en una topología. Los spouts generalmente leerán tuplas desde una fuente externa y las emiten dentro de la topología.

Un bolt es donde se realiza todo el procesamiento de una topología, pueden hacer cualquier cosa, desde filtrado, funciones, agregaciones, joins, comunicarse con bases de datos y mucho más.

El ejemplo que hice y compartiré a continuación con ustedes será bastante simple, estará constituido por un spout que leerá de una cola rabbitmq (la cola data que creamos anteriormente) y ese mensaje lo insertará en la topología para posteriormente al recibirlo el bolt mostrarlo por linea de comandos (ya luego si ustedes lo desean lo que podrían hacer es que en vez de mostrarlo por linea de comandos volcar ese mensaje en otra cola de rabbitmq).

El programa java que se encarga de insertar mensajes a la cola

package com.josedeveloper.rabbitmq;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class SendMessagesToRabbit {

    public static final String message = "MENSAJE DE PRUEBA";
    public static final int NUM_MENSAJES = 10000;

    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
		sendMessage();
    }

    public static void sendMessage() throws IOException, InterruptedException, TimeoutException {
    	ConnectionFactory factory = new ConnectionFactory();
	    factory.setHost("localhost");
	    factory.setPort(5672);
	    factory.setUsername("guest"); //usuario por defecto de rabbitmq
	    factory.setPassword("guest"); //password por defecto de rabbitmq
	    factory.setVirtualHost("/");

	    Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();

		for (int i = 0; i &lt; NUM_MENSAJES; i++) {
			final String msg = message + i;
			channel.basicPublish("", "data", null, msg.getBytes());
		}


		channel.close();
	    connection.close();
    }

}

package com.josedeveloper.rabbitmq;

import java.io.IOException;

import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

public class SendMessagesToRabbit {

public static final String message = "MENSAJE DE PRUEBA";

public static final int NUM_MENSAJES = 10000;

public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {

sendMessage();

}

public static void sendMessage() throws IOException, InterruptedException, TimeoutException {

ConnectionFactory factory = new ConnectionFactory();

factory.setHost("localhost");

factory.setPort(5672);

factory.setUsername("guest"); //usuario por defecto de rabbitmq

factory.setPassword("guest"); //password por defecto de rabbitmq

factory.setVirtualHost("/");

Connection connection = factory.newConnection();

Channel channel = connection.createChannel();

for (int i = 0; i < NUM_MENSAJES; i++) {

final String msg = message + i;

channel.basicPublish("", "data", null, msg.getBytes());

}

channel.close();

connection.close();

}

El Spout que leerá de la cola rabbitMQ

package com.josedeveloper.topologia;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

public class RabbitMQSpout extends BaseRichSpout {

	private static final long serialVersionUID = -5875062340173997062L;

	private SpoutOutputCollector collector;
	BlockingQueue messages;

	private final static String QUEUE_NAME = "data";

	@Override
	public void nextTuple() {
		String message;
        while ((message = messages.poll()) != null) {
        	collector.emit(new Values(message)); //emitimos el mensaje dentro de la topologia
        }

	}

	@Override
	public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
		this.collector = collector;
		messages = new ArrayBlockingQueue(100);
		ConnectionFactory factory = new ConnectionFactory();
	    factory.setHost("localhost");
	    Connection connection;

	    try {
			connection = factory.newConnection();
			Channel channel = connection.createChannel();

			channel.queueDeclare(QUEUE_NAME, true, false, false, null);

			Consumer consumer = new DefaultConsumer(channel) {
			      @Override
			      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
			          throws IOException {
			        String message = new String(body, "UTF-8");
			        try {
						messages.put(message);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
			      }
			    };
		    channel.basicConsume(QUEUE_NAME, true, consumer);
		} catch (IOException e) {
			e.printStackTrace();
		} catch (TimeoutException e1) {
			e1.printStackTrace();
		}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare( new Fields( "message" ) ); //declaramos los campos que enviaremos a la topologia
	}

}

package com.josedeveloper.topologia;

import java.io.IOException;

import java.util.Map;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.TimeoutException;

import org.apache.storm.spout.SpoutOutputCollector;

import org.apache.storm.task.TopologyContext;

import org.apache.storm.topology.OutputFieldsDeclarer;

import org.apache.storm.topology.base.BaseRichSpout;

import org.apache.storm.tuple.Fields;

import org.apache.storm.tuple.Values;

import com.rabbitmq.client.AMQP;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

import com.rabbitmq.client.Consumer;

import com.rabbitmq.client.DefaultConsumer;

import com.rabbitmq.client.Envelope;

public class RabbitMQSpout extends BaseRichSpout {

private static final long serialVersionUID = -5875062340173997062L;

private SpoutOutputCollector collector;

BlockingQueue messages;

private final static String QUEUE_NAME = "data";

@Override

public void nextTuple() {

String message;

while ((message = messages.poll()) != null) {

collector.emit(new Values(message)); //emitimos el mensaje dentro de la topologia

}

@Override

public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {

this.collector = collector;

messages = new ArrayBlockingQueue(100);

ConnectionFactory factory = new ConnectionFactory();

factory.setHost("localhost");

Connection connection;

try {

connection = factory.newConnection();

Channel channel = connection.createChannel();

channel.queueDeclare(QUEUE_NAME, true, false, false, null);

Consumer consumer = new DefaultConsumer(channel) {

@Override

public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)

throws IOException {

String message = new String(body, "UTF-8");

try {

messages.put(message);

} catch (InterruptedException e) {

e.printStackTrace();

}

};

channel.basicConsume(QUEUE_NAME, true, consumer);

} catch (IOException e) {

e.printStackTrace();

} catch (TimeoutException e1) {

e1.printStackTrace();

}

@Override

public void declareOutputFields(OutputFieldsDeclarer declarer) {

declarer.declare( new Fields( "message" ) ); //declaramos los campos que enviaremos a la topologia

}

El bolt que leerá los datos que han sido insertados en la topología por el Spout

package com.josedeveloper.topologia;

import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

public class MessageBolt extends BaseRichBolt {

	private static final long serialVersionUID = 1L;

	@SuppressWarnings("unused")
	private OutputCollector collector;

	@Override
	public void execute(Tuple tuple) {
		String message = tuple.getString(0);
		System.out.println("--&gt; " + message);
	}

	@Override
	public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer arg0) {
		// TODO Auto-generated method stub

	}

}

package com.josedeveloper.topologia;

import java.util.Map;

import org.apache.storm.task.OutputCollector;

import org.apache.storm.task.TopologyContext;

import org.apache.storm.topology.OutputFieldsDeclarer;

import org.apache.storm.topology.base.BaseRichBolt;

import org.apache.storm.tuple.Tuple;

public class MessageBolt extends BaseRichBolt {

private static final long serialVersionUID = 1L;

@SuppressWarnings("unused")

private OutputCollector collector;

@Override

public void execute(Tuple tuple) {

String message = tuple.getString(0);

System.out.println("--> " + message);

}

@Override

public void prepare(Map conf, TopologyContext context, OutputCollector collector) {

this.collector = collector;

}

@Override

public void declareOutputFields(OutputFieldsDeclarer arg0) {

// TODO Auto-generated method stub

}

Definición de la topología

package com.josedeveloper.topologia;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.Utils;

public class RabbitMQTopologyExample {

	public static void main(String[] args) {
		TopologyBuilder builder = new TopologyBuilder();

        builder.setSpout("spout", new RabbitMQSpout());
        builder.setBolt("bolt", new MessageBolt())
                .shuffleGrouping("spout");

        Config conf = new Config();

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("test", conf, builder.createTopology());
        Utils.sleep(200000);
        cluster.killTopology("test");
        cluster.shutdown();
	}

}

package com.josedeveloper.topologia;

import org.apache.storm.Config;

import org.apache.storm.LocalCluster;

import org.apache.storm.topology.TopologyBuilder;

import org.apache.storm.utils.Utils;

public class RabbitMQTopologyExample {

public static void main(String[] args) {

TopologyBuilder builder = new TopologyBuilder();

builder.setSpout("spout", new RabbitMQSpout());

builder.setBolt("bolt", new MessageBolt())

.shuffleGrouping("spout");

Config conf = new Config();

LocalCluster cluster = new LocalCluster();

cluster.submitTopology("test", conf, builder.createTopology());

Utils.sleep(200000);

cluster.killTopology("test");

cluster.shutdown();

}

Para ejecutar nuestro ejemplo y verlo funcionando debemos ejecutar 2 clases (el orden sería indistinto):

SendMessagesToRabbit
RabbitMQTopologyExample

Al ejecutar la clase SendMessagesToRabbit, podremos ver en el cliente Web de RabbitMQ como la cola tendrá 10000 mensajes encolados. Al ejecutar la topología (ejecutando la clase RabbitMQTopologyExample) podremos ver como los mensajes se van desencolando y a su vez por linea de comandos (por ejemplo de nuestro editor) veremos los mensajes que en teoría el bolt leyó y procesó.

Espero que les sea de utilidad y disfruten con este framework, desde mi punto de vista es sencillo y funciona bien, incluso la nueva herramienta que utiliza Twitter posee retrocompatibilidad con Storm por lo cual se podría empezar con un ejemplo de este tipo.

Clic aquí para ir al repositorio github.

Sabias que Java posee una clase Void

Posted on May 27, 2016 by admin

Hacía mas de 2 años que quería hacer este post, porque me parece algo bastante curioso. De hecho fue en el 2012 que me enteré de la existencia de la clase java.lang.Void pero lo verdaderamente interesante es que esta clase existe desde la versión 1.1 del jdk.

De acuerdo a la documentación de la API “La clase Void es un comodín para almacenar una referencia a objetos de clases representando la palabra clave void.”

A continuación un ejemplo que muestran el uso de la clase Void. La primera parte es con Generics (que de hecho fue así como me enteré de su existencia), lo que me resulta curioso es que quizás ahora este sea su uso mas extendido aun cuando Generics apareció en la versión 5 de Java. La segunda parte es con reflection el cual creo que fue el primer caso de uso de esta clase Void.

package com.josedeveloper.example;

interface Ejecutable  {

	public T ejecutar();

}

class SumaEjemplo implements Ejecutable {

	int a;
	int b;

	public SumaEjemplo(int a, int b) {
		this.a = a;
		this.b = b;
	}

	public Integer ejecutar() {
		return a + b;
	}

}

class PrintEjemplo implements Ejecutable {

	public Void ejecutar() {
		System.out.println("Esta clase no devuelve nada");
		return null;
	}

	public void prueba() {
		System.out.println("Prueba");
	}

}

public class VoidExample {

	public static void main(String[] args) throws NoSuchMethodException, SecurityException {

		Ejecutable suma = new SumaEjemplo(10, 15);
		System.out.println("Resultado: " + suma.ejecutar());

		Ejecutable print = new PrintEjemplo();
		print.ejecutar();

		//Ejemplo de reflection
		PrintEjemplo ejemploReflection = new PrintEjemplo();

		if (ejemploReflection.getClass().getMethod("prueba").getReturnType() == Void.TYPE)
			System.out.println("el es del tipo void");
	}

}

package com.josedeveloper.example;

interface Ejecutable {

public T ejecutar();

}

class SumaEjemplo implements Ejecutable {

int a;

int b;

public SumaEjemplo(int a, int b) {

this.a = a;

this.b = b;

}

public Integer ejecutar() {

return a + b;

}

class PrintEjemplo implements Ejecutable {

public Void ejecutar() {

System.out.println("Esta clase no devuelve nada");

return null;

}

public void prueba() {

System.out.println("Prueba");

}

public class VoidExample {

public static void main(String[] args) throws NoSuchMethodException, SecurityException {

Ejecutable suma = new SumaEjemplo(10, 15);

System.out.println("Resultado: " + suma.ejecutar());

Ejecutable print = new PrintEjemplo();

print.ejecutar();

//Ejemplo de reflection

PrintEjemplo ejemploReflection = new PrintEjemplo();

if (ejemploReflection.getClass().getMethod("prueba").getReturnType() == Void.TYPE)

System.out.println("el es del tipo void");

}

Tablas en memoria en MySQL

Posted on March 24, 2016 by admin

Algunas semanas atrás en el trabajo nos topamos con un problema, realizábamos una consulta que consistía en hacer un join entre una tabla muy grande contra un par de tablas pequeñas que contenían únicamente descripciones. Como ustedes ha de imaginarse teníamos nuestra base de datos normalizada y para exportar los datos con el conjunto de descripciones era necesario realizar este cruce.

Para que la consulta fuese bien (que no arrojase un timeout en el mejor de los casos) habíamos agregado índices a las tablas secundarias (aquellas que contenían las descripciones) y la consulta de lograr ejecutarse llegó a tardar unos 25 segundos. Estas tablas eran de tipo InnoDB y los índices eran de tipo B-Tree (es el único tipo de índice para las tablas de tipo InnoDB). La consulta llegaba a tardar en el mejor de los casos (que no devolviese un timeout) 25 segundos. Es por esta razón que buscando una alternativa abordamos las tablas en memoria.

Las tablas en memoria son almacenadas en la memoria HEAP razón por la cual son tablas temporales ya que una vez se haya reiniciado o apagado la máquina estas perderán los datos, es decir se mantiene la estructura de la tabla mas no su contenido. Las tablas en memoria en MySQL pueden tener 2 tipos de índices B-Tree o Hash.

¿Cuando utilizar cada tipo de índice?

El índice B-Tree puede utilizarse con operadores del tipo =, >, >=, <, <= o BETWEEN, de igual manera puede ser usado con el operador LIKE, mientras que el índice Hash es usado únicamente para comparaciones de igualdad como =, >= o <=. Así que de realizar comparaciones de igualdad es recomendable utilizar el índice Hash ya que es muy rápido mientras que cuando se realicen comparaciones del tipo >, <, LIKE, BETWEEN u optimizar operaciones de ORDER BY la opción es utilizar B-Tree.

En nuestro caso únicamente realizábamos operaciones de igualdad ya que comparábamos en base a un código así que utilizamos índices de tipo Hash, obteniendo excelentes resultados, tanto es así que nunca llegamos a obtener un timeout y el tiempo de respuesta de nuestras consultas se redujeron a 5 segundos.

¿Cómo crear una tabla en memoria?

La creación de una tabla en memoria es muy sencilla, es igual a la creación de cualquier otra tabla del tipo InnoDB, lo único es sencillamente cambiar el tipo de “engine” e indicar “MEMORY”, por ejemplo:

CREATE TABLE `PRODUCTO_MEM` (
`CODIGO` int(11) NOT NULL,
`DESCRIPCION` varchar(45) NOT NULL,
KEY `index1` (`CODIGO`) USING HASH
) ENGINE=MEMORY DEFAULT CHARSET=latin1;

CREATE TABLE `PRODUCTO_MEM` (

`CODIGO` int(11) NOT NULL,

`DESCRIPCION` varchar(45) NOT NULL,

KEY `index1` (`CODIGO`) USING HASH

) ENGINE=MEMORY DEFAULT CHARSET=latin1;

Por último ¿Como hacer para que se carguen los datos de la tabla en memoria al arrancar?

Por muy tonto que parezca para nosotros esto era importante ya que nos asegurábamos de que de reiniciarse el ordenador (era windows :-s ) estando o no nosotros, las tablas en memoria estarían creadas y cualquier consulta que se efectuase no daría error. Para que la tabla en memoria se llene de datos al arrancar el servidor MySQL es bastante sencillo, lo único que debemos hacer es editar el fichero my.ini o my.cnf y debajo de donde dice [mysqld] agregar la siguiente linea

...

[mysqld]

init-file="/Users/my_home/script.sql"

...

...

[mysqld]

init-file="/Users/my_home/script.sql"

...

Donde script.sql tendrá las sentencias sql de inserción por ejemplo

INSERT INTO `my_db`.`PRODUCTO_MEM` (`CODIGO`,`DESCRIPCION`) VALUES (1, 'descripcion del producto 1');
INSERT INTO `my_db`.`PRODUCTO_MEM` (`CODIGO`,`DESCRIPCION`) VALUES (2, 'descripcion del producto 2');

1 2	INSERT INTO `my_db`.`PRODUCTO_MEM` (`CODIGO`,`DESCRIPCION`) VALUES (1, 'descripcion del producto 1'); INSERT INTO `my_db`.`PRODUCTO_MEM` (`CODIGO`,`DESCRIPCION`) VALUES (2, 'descripcion del producto 2');

Espero que esto les pueda servir para optimizar el consultas en base de datos o sencillamente resolver cualquier inconveniente.

¿Has trabajado antes con tablas en memoria? Compartirías con nosotros tu experiencia.

Web scraping con Java

Posted on July 25, 2015 by admin

Actualmente me encuentro desarrollando mi proyecto final de máster, el cual consiste en crear un modelo de aprendizaje automático que arroje predicciones acerca de partidos de futbol de la liga de primera división española. Para ello he necesitado entre otras cosas tener los resultados de todas las jornadas de las ultimas ligas. Aunque recientemente conseguí un paquete de R (enlace) que contenía los resultados desde 1929, este no me proporcionaba toda la información que yo buscaba, así que me decidí por obtener yo mismo esa información sacándola de las paginas deportivas y es lo que quiero compartir con ustedes.

En un principio pense en hacerlo en python con la biblioteca lxml, pero haciendo una búsqueda rápida por Internet encontré un proyecto en Java llamado Jsoup y debo decir que este si me simplifico la tarea.

Primero como todos saben es necesario que demos un repaso a la estructura del documento que vamos a scrapear y confirmar que hay un patron.

Como pueden ver en la imagen, podemos detallar que todas las filas de las tablas de las jornadas comparten el atributo itemtype=”http://schema.org/SportEvent”, así que este fue el que utilicé para obtener todas las filas y a partir de allí obtener los nombres de los equipos, el resultado y el enlace para ir al detalle del partido.

package com.josedeveloper.WebScrapingExample;

import java.io.IOException;

import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;

public class App
{
    public static void main( String[] args ) throws IOException
    {
    	final String url = "http://resultados.as.com/resultados/futbol/primera/2014_2015/calendario";

    	Document doc = Jsoup.connect(url).get();

    	//Obtenemos todas las filas identificadas como evento deportivo
    	//ya que con este atributo es como se identifican los partidos
    	Elements matches = doc.select("tr[itemtype$=\"http://schema.org/SportsEvent\"]");

    	for (Element match: matches) {

    		//Obtenemos los equipos de cada partido utilizando también expresiones
    		Elements teams = match.select("td[itemtype$=\"http://schema.org/SportsTeam\"]");

    		//obtenemos el enlace al detalle del partido
    		Elements score = match.select("a[class=\"resultado resul_post\"]");

    		String localTeam = teams.get(0).text();
    		String visitorTeam = teams.get(1).text();
    		String statsLink = score.first().attr("href");

    		String[] goals = score.first().text().split("-");
    		int localGoals = Integer.parseInt(goals[0].trim());
    		int visitorGoals = Integer.parseInt(goals[1].trim());

    		System.out.println(localTeam + " vs " + visitorTeam + ": " + localGoals + "-" + visitorGoals + " -&gt; " + statsLink);
    	}
    }
}

package com.josedeveloper.WebScrapingExample;

import java.io.IOException;

import org.jsoup.Jsoup;

import org.jsoup.nodes.Document;

import org.jsoup.nodes.Element;

import org.jsoup.select.Elements;

public class App

{

public static void main( String[] args ) throws IOException

{

final String url = "http://resultados.as.com/resultados/futbol/primera/2014_2015/calendario";

Document doc = Jsoup.connect(url).get();

//Obtenemos todas las filas identificadas como evento deportivo

//ya que con este atributo es como se identifican los partidos

Elements matches = doc.select("tr[itemtype$=\"http://schema.org/SportsEvent\"]");

for (Element match: matches) {

//Obtenemos los equipos de cada partido utilizando también expresiones

Elements teams = match.select("td[itemtype$=\"http://schema.org/SportsTeam\"]");

//obtenemos el enlace al detalle del partido

Elements score = match.select("a[class=\"resultado resul_post\"]");

String localTeam = teams.get(0).text();

String visitorTeam = teams.get(1).text();

String statsLink = score.first().attr("href");

String[] goals = score.first().text().split("-");

int localGoals = Integer.parseInt(goals[0].trim());

int visitorGoals = Integer.parseInt(goals[1].trim());

System.out.println(localTeam + " vs " + visitorTeam + ": " + localGoals + "-" + visitorGoals + " -> " + statsLink);

}

Par de cosas que quisiera comentar con respecto al código:

Podemos aplicar expresiones sobre elementos de antemano obtenidos, por ejemplo como se hizo para obtener los equipos que intervienen en el partido.
Existe otra forma además de la anteriormente explicada (usando expresiones) para obtener elementos del árbol DOM de la página Web, si damos un vistazo a la API de la biblioteca del partido, existe un método getElementsByAttributeValue, entonces para obtener el elemento score, este se pudo haber obtenido también de la siguiente manera
Elements score = match.getElementsByAttributeValue(“class”, “resultado resul_post”)
Por último si quisiéramos obtener mas datos por ejemplo del detalle del partido (ya que logramos obtener el url), esta biblioteca nos permite seguir navegando (haciendo conexiones), y sería cuestión de realizar otra conexión y de nuevo empezar a extraer elementos.
Document detalleDelPartido = Jsoup.connect(statsLink).get()

Aquí les dejo el enlace al repositorio GitHub y espero que les pueda ser de utilidad.

Estadística simple con Spark

Posted on July 11, 2015 by admin

Hace unos 20 días lei un articulo titulado “Simple Data Analysis Using Spark” (no publico el enlace ya que el mismo desapareció de dzone de la zona de Big Data y el mismo blog ya no existe, corroborarlo buscando por Internet), lo interesante de este articulo es que el autor hacía cálculos de estadística simple, es decir, calculaba, media, máximo y mínimo, y fue casualidad que justo en ese momento acaba de leer acerca de las operaciones numéricas de los RDD, es esta la razón que me movió a querer hacer un ejercicio similar, con estadística simple pero quería hacerlo con una información que yo considerara valiosa.

Fue entonces que me decidí a buscar alguna fuente de open data española, ya que quería arrojar u obtener datos reales, o no se quizás quería sencillamente obtener algo significativo que se acercara a una tarea real, fue así que llegue al Portal de datos abiertos de la comunidad de Madrid y después de revisar el catalogo me decidí por el padrón municipal, un fichero csv de 22 MB.

He aquí el código:

package com.josedeveloper

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._

case class Padron(codDistrito: String, descDistrito: String, codDistBarrio: String,
             descBarrio: String, codBarrio: String, codDistSeccion: String, spainMen: Integer,
             spainWomen: Integer, foreignerMen: Integer, foreignerWomen: Integer) {

  def numberOfPeople() : Int = {
    return spainMen + spainWomen + foreignerMen + foreignerWomen;
  }

}

object ScalaApp extends App {

  val sc = new SparkContext("local", "Simple Application", "$SPARK_HOME"
    , List("target/SimpleApp-0.1.jar"))

  val file = sc.textFile("Rango_Edades_Seccion_201506.csv")

  val data = file.map(line =&gt; line.split(";").map(_.trim))
      //the csv header is excluded
      .mapPartitionsWithIndex((idx, iter) =&gt; if (idx == 0) iter.drop(1) else iter).persist()

  //Map to Padron objects
  val dataMapped = data.map(line =&gt; (line(1), new Padron(line(0), line(1),
    line(2), line(3), line(4), line(5),
    getIntValue(line(8)),
    getIntValue(line(9)),
    getIntValue(line(10)),
    getIntValue(line(11)))))

  //Groupped by district
  val grouppedData = dataMapped.reduceByKey((x: Padron, y: Padron) =&gt; new Padron(x.codDistrito, x.descDistrito,
    x.codDistBarrio, x.descBarrio, x.codBarrio, x.codDistSeccion, x.spainMen + y.spainMen,
    x.spainWomen + y.spainWomen, x.foreignerMen + y.foreignerMen,
    x.foreignerWomen + y.foreignerWomen))

  //sorted list total spanish men by district
  grouppedData.collect().sortBy(_._2.spainMen)
    .foreach(x =&gt; println(x._2.descDistrito + "-&gt;" + x._2.spainMen))

  //statistics and numerics sparks RDD values
  val spainMaleValeByDistrict = grouppedData.map(tupla =&gt; tupla._2.spainMen.doubleValue()).cache()
  val media = spainMaleValeByDistrict.mean()
  val stddev = spainMaleValeByDistrict.stdev()
  val max = spainMaleValeByDistrict.max()
  val min = spainMaleValeByDistrict.min()

  println("Media de Españoles varones por distrito: " + media.toInt)
  println("Desviación estandar de españoles varones por distrito: " + stddev.toInt)
  println("Num maximo de españoles varones en un distrito: " + max.toInt)
  println("Num minimo de españoles varones en un distrito: " + min.toInt)


  //sorted list total number of people by district
  grouppedData.collect().sortBy(_._2.numberOfPeople)
    .foreach(x =&gt; println(x._2.descDistrito + "-&gt;" + x._2.numberOfPeople))

  val numberOfPeopleByDistrict = grouppedData.map(tupla =&gt; tupla._2.numberOfPeople.doubleValue).cache()
  val media2 = numberOfPeopleByDistrict.mean()
  val stddev2 = numberOfPeopleByDistrict.stdev()
  val max2 = numberOfPeopleByDistrict.max()
  val min2 = numberOfPeopleByDistrict.min()

  println("Media de personas por distrito: " + media2.toInt)
  println("Desviación estandar de personas por distrito: " + stddev2.toInt)
  println("Num maximo de personas en un distrito: " + max2.toInt)
  println("Num minimo de personas en un distrito: " + min2.toInt)

  def getIntValue(s:String) : Integer = {
    val value = s.substring(1, s.length - 1)
    if (value.isEmpty)
      return 0
    return Integer.parseInt(value)
  }

}

package com.josedeveloper

import org.apache.spark.SparkContext

import org.apache.spark.SparkContext._

case class Padron(codDistrito: String, descDistrito: String, codDistBarrio: String,

descBarrio: String, codBarrio: String, codDistSeccion: String, spainMen: Integer,

spainWomen: Integer, foreignerMen: Integer, foreignerWomen: Integer) {

def numberOfPeople() : Int = {

return spainMen + spainWomen + foreignerMen + foreignerWomen;

}

object ScalaApp extends App {

val sc = new SparkContext("local", "Simple Application", "$SPARK_HOME"

, List("target/SimpleApp-0.1.jar"))

val file = sc.textFile("Rango_Edades_Seccion_201506.csv")

val data = file.map(line => line.split(";").map(_.trim))

//the csv header is excluded

.mapPartitionsWithIndex((idx, iter) => if (idx == 0) iter.drop(1) else iter).persist()

//Map to Padron objects

val dataMapped = data.map(line => (line(1), new Padron(line(0), line(1),

line(2), line(3), line(4), line(5),

getIntValue(line(8)),

getIntValue(line(9)),

getIntValue(line(10)),

getIntValue(line(11)))))

//Groupped by district

val grouppedData = dataMapped.reduceByKey((x: Padron, y: Padron) => new Padron(x.codDistrito, x.descDistrito,

x.codDistBarrio, x.descBarrio, x.codBarrio, x.codDistSeccion, x.spainMen + y.spainMen,

x.spainWomen + y.spainWomen, x.foreignerMen + y.foreignerMen,

x.foreignerWomen + y.foreignerWomen))

//sorted list total spanish men by district

grouppedData.collect().sortBy(_._2.spainMen)

.foreach(x => println(x._2.descDistrito + "->" + x._2.spainMen))

//statistics and numerics sparks RDD values

val spainMaleValeByDistrict = grouppedData.map(tupla => tupla._2.spainMen.doubleValue()).cache()

val media = spainMaleValeByDistrict.mean()

val stddev = spainMaleValeByDistrict.stdev()

val max = spainMaleValeByDistrict.max()

val min = spainMaleValeByDistrict.min()

println("Media de Españoles varones por distrito: " + media.toInt)

println("Desviación estandar de españoles varones por distrito: " + stddev.toInt)

println("Num maximo de españoles varones en un distrito: " + max.toInt)

println("Num minimo de españoles varones en un distrito: " + min.toInt)

//sorted list total number of people by district

grouppedData.collect().sortBy(_._2.numberOfPeople)

.foreach(x => println(x._2.descDistrito + "->" + x._2.numberOfPeople))

val numberOfPeopleByDistrict = grouppedData.map(tupla => tupla._2.numberOfPeople.doubleValue).cache()

val media2 = numberOfPeopleByDistrict.mean()

val stddev2 = numberOfPeopleByDistrict.stdev()

val max2 = numberOfPeopleByDistrict.max()

val min2 = numberOfPeopleByDistrict.min()

println("Media de personas por distrito: " + media2.toInt)

println("Desviación estandar de personas por distrito: " + stddev2.toInt)

println("Num maximo de personas en un distrito: " + max2.toInt)

println("Num minimo de personas en un distrito: " + min2.toInt)

def getIntValue(s:String) : Integer = {

val value = s.substring(1, s.length - 1)

if (value.isEmpty)

return 0

return Integer.parseInt(value)

}

Hay una cosa que quiero resaltar, Lo simple y corto del código en Scala, tanto para la definición de la clase Padron como en las transformaciones hechas en los RDD, ya Scala esta en mi lista de cosas por aprender porque se que con Java hubiese sido el doble de código.

Por último les dejo parte de la información obtenida, la lista ordenada de población por distrito:

BARAJAS: 46166
VICALVARO: 69709
MORATALAZ: 94785
VILLA DE VALLECAS: 101011
MONCLOA-ARAVACA: 116581
RETIRO: 118390
CENTRO: 131805
USERA: 133579
CHAMBERI: 137454
VILLAVERDE: 141457
CHAMARTIN: 142334
SALAMANCA: 143123
ARGANZUELA: 151061
TETUAN: 152142
SAN BLAS-CANILLEJAS: 153303
HORTALEZA: 176320
CIUDAD LINEAL: 212626
PUENTE DE VALLECAS: 227183
LATINA: 234842
FUENCARRAL-EL PARDO: 234883
CARABANCHEL: 242032

¿Te ha parecido interesante? ¿Qué le agregarías o quitarías al código?

Si quieres acceder al GitHub donde he colgado el código pincha aquí

Primer ejemplo con apache spark

Posted on June 12, 2015 by admin

Hace algunas semanas atrás que empezamos a trabajar con apache spark en el máster, brevemente les contare mis impresiones desde mi punto de vista como principiante.

Apache spark me gusto, ¿por qué? porque se puede programar en Scala (además de Python y Java), la API de Scala lo simplifica mucho y la cantidad de código a teclear es (considerablemente) menor a la necesaria para hacer la misma tarea en Java, aunque aquí debo hacer un paréntesis, ya que con la entrada de Java 8 y las lambda expresión la cantidad de código será menor pero insisto la API de Scala a mi modo de ver lo hace mas sencillo.

Nos permite hacer operaciones con mucha data (quizás no Big Data, es decir no hablamos de TeraBytes) sin necesidad de usar Hadoop, me gustaría hacer hincapié en esto ya que muchos piensan que Big Data es Hadoop y NO, no es así, mi concepto de Big Data es un poco mas amplio y contempla mas cosas, aquí me refiero específicamente a la API MapReduce de Hadoop la cual es mas complicada de usar que la de Spark, no obstante si necesitamos trabajar con una cantidad de datos considerable que requiera del uso de HDFS, podemos desde Spark acceder a este sistema de archivo u algún otro como por ejemplo Amazon S3.

Por último decirles que Spark es mas rapido, esto se debe a su arquitectura, ya que este trabaja en memoria (todo en memoria si tiene la capacidad de cargar todos los datos por completo) y pues si llegamos a hacer varias tareas (lo que técnicamente sería reutilizar los RDD ya guardados en memoria) con los datos en memoria es cuando mas jugo se le sacará a Spark.

Sinceramente si alguien discrepa de mi o coincide me entusiasmaría mucho que comparta su opinión ya que todo este mundo que envuelve el Big Data me tiene super enganchado.

Ahora bien vamos con los ejemplos, estas formaban parte de la primera tarea de Spark que tuvimos que realizar, y consiste en 2 ejemplos:

Dado un fichero de texto en formato csv con información de medallistas olímpicos, obtener el numero de medallas por edad.
Dado el mismo fichero de texto anterior, y un esquema de puntuación por tipo de medalla obtener un ranking de medallistas olímpicos.

He aqui el código:

package org.dummy

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._

object ScalaApp extends App {
  val logFile = "OlympicAthletes.csv"
  val sc = new SparkContext("local", "Simple", "$SPARK_HOME"
    , List("target/spark-ejercicio-1.0.jar"))

  //Todo el fichero csv es cargado en este RDD
  val file = sc.textFile(logFile)

  //linea a linea se aplica un split o separacion utilizando la ","
  //eso nos arrojara un arreglo de palabras (o numeros)
  //y construimos una instancia del tipo OlympicMedalRecords
  val olympicMedalRecordsRDD = file.map(x =&gt; {
    val arr = x.split(",")
    new OlympicMedalRecords(arr(0), Integer.parseInt(arr(1)), arr(2)
      , Integer.parseInt(arr(3)), arr(5), Integer.parseInt(arr(6)),
      Integer.parseInt(arr(7)), Integer.parseInt(arr(8)))
  }
  )


  //Mis ejercicios
  //Ejercicio 1
  //mapeo del tipo (edad, num de medallas), val medal es un RDD
  val medal = olympicMedalRecordsRDD.map(record =&gt; (record.getAge, record.getGoldMedals + record.getSilverMedals + record.getBronzeMedals))

  //creamos un nuevo RDD el cual manejara la suma de todas las madellas por edad
  val sum = medal.reduceByKey((acc: Int,value: Int)=&gt;acc+value)
  println("Lista de Medallas ordenado por Edad")

  //ordenamos, juntamos y luego recorremos para mostrar todos los valores
  sum.sortBy(_._1).collect.foreach(println)


  //Ejercicio 2 hacer un ranking por atleta
  // ya que oro = 3 ptos, plata = 2ptos, bronce = 1pto
  //sportGuy es otro RDD y aqui hacemos un mapeo del tipo (nombre_atleta, puntos_por_medalla)
  val sportGuy = olympicMedalRecordsRDD.map(record =&gt; (record.getName, 3*record.getGoldMedals + 2*record.getSilverMedals + record.getBronzeMedals))

  //ranking es a su vez otro RDD  y de igual manera como se hizo en el ejemplo anterior se agrupa por nombre de atleta
  // y también utilizamos una variable acumuladora acc donde vamos sumando los ptos de un determinado atleta
  val ranking = sportGuy.reduceByKey((acc,value)=&gt;acc+value)
  println("Ranking de deportistas")
  // ordenamos en sentido inverso y mostramos  los atletas con mas ptos
  ranking.sortBy(_._2, false).collect.foreach(println)
}

package org.dummy

import org.apache.spark.SparkContext

import org.apache.spark.SparkContext._

object ScalaApp extends App {

val logFile = "OlympicAthletes.csv"

val sc = new SparkContext("local", "Simple", "$SPARK_HOME"

, List("target/spark-ejercicio-1.0.jar"))

//Todo el fichero csv es cargado en este RDD

val file = sc.textFile(logFile)

//linea a linea se aplica un split o separacion utilizando la ","

//eso nos arrojara un arreglo de palabras (o numeros)

//y construimos una instancia del tipo OlympicMedalRecords

val olympicMedalRecordsRDD = file.map(x => {

val arr = x.split(",")

new OlympicMedalRecords(arr(0), Integer.parseInt(arr(1)), arr(2)

, Integer.parseInt(arr(3)), arr(5), Integer.parseInt(arr(6)),

Integer.parseInt(arr(7)), Integer.parseInt(arr(8)))

}

)

//Mis ejercicios

//Ejercicio 1

//mapeo del tipo (edad, num de medallas), val medal es un RDD

val medal = olympicMedalRecordsRDD.map(record => (record.getAge, record.getGoldMedals + record.getSilverMedals + record.getBronzeMedals))

//creamos un nuevo RDD el cual manejara la suma de todas las madellas por edad

val sum = medal.reduceByKey((acc: Int,value: Int)=>acc+value)

println("Lista de Medallas ordenado por Edad")

//ordenamos, juntamos y luego recorremos para mostrar todos los valores

sum.sortBy(_._1).collect.foreach(println)

//Ejercicio 2 hacer un ranking por atleta

// ya que oro = 3 ptos, plata = 2ptos, bronce = 1pto

//sportGuy es otro RDD y aqui hacemos un mapeo del tipo (nombre_atleta, puntos_por_medalla)

val sportGuy = olympicMedalRecordsRDD.map(record => (record.getName, 3*record.getGoldMedals + 2*record.getSilverMedals + record.getBronzeMedals))

//ranking es a su vez otro RDD y de igual manera como se hizo en el ejemplo anterior se agrupa por nombre de atleta

// y también utilizamos una variable acumuladora acc donde vamos sumando los ptos de un determinado atleta

val ranking = sportGuy.reduceByKey((acc,value)=>acc+value)

println("Ranking de deportistas")

// ordenamos en sentido inverso y mostramos los atletas con mas ptos

ranking.sortBy(_._2, false).collect.foreach(println)

}

Una última cosa que es bueno saber para aquellos que como yo están empezando con Spark y sus RDD, Los RDD tienen 2 tipos de operaciones: transformaciones y acciones, las primeras son operaciones que como resultado nos devuelven otro RDD y las acciones son aquellas que hacen “algo” sobre los datos como por ejemplo el foreach de mi código, utilizado para mostrar por pantalla los valores finales, además los RDD (como hasta ahora lo entiendo) son las instrucciones a realizar sobre los datos, mas no almacenan la información, además estás instrucciones no se realizan de inmediato, sino que al contrario estos tienen un funcionamiento lazy, esto significa que se ejecutan cada vez que se realiza una operación de tipo acción.

Otra sugerencia, no duden en cacharrear con la consola de spark, de hecho el primer ejercicio lo llegue a hacer por la consola (sin los JavaBeans) y el practicar por aquí si que ayuda a conocer los operaciones a utilizar sobre los RDD.

Aquellos interesados en descargar el código, les dejo la dirección del ejercicio en mi repositorio github

El esqueleto del ejercicio esta en el repositorio del que fue nuestro profesor y pueden acceder a el haciendo clic aquí.

Cualquier material, idea o contenido que quieres compartir, no dudes en hacerlo, ya que bien recibido será. Ahora si para cerrar les dejo el enlace a un libro que hasta ahora me esta pareciendo muy bueno para aprender Spark (además con ejemplos elaborados en Java, Scala y Python).

Qué es Big Data

Posted on June 12, 2015 by admin

Hace una semana en una clase del máster mas enfocada a negocio, retos y oportunidades del Big Data, hicimos un ejercicio bastante interesante, este consistió en que cada uno de nosotros brevemente plasmara su definición de Big Data, donde podría aplicarse y que beneficios podría arrojarnos.

Pudimos ver las definiciones de algunos compañeros, coincidir y discrepar en algunos puntos pero para mí fue una actividad muy enriquecedora sin importar si coincidíamos o no, es por eso que comparto con ustedes mi definición de Big Data.

¿Qué es Big Data?

Big Data contempla las nuevas herramientas, tecnologías y (nuevos) los conceptos relacionados con la adquisición de (mucha) data (volumen), de distinto tipo (variedad) que a su vez podría estar no estructurada, con unos aspectos opcionales pero que también puede marcar la diferencia para definirlo como “really Big Data” como la movilidad (por ejemplo la adquisición de información mediante IoT o dispositivos móviles) y el tiempo real. De hecho contemplo el Big Data como unas 2 vertientes que pueden o no trabajar en conjunto:

Ingeniería: Enfocado en el uso de las herramientas por ejemplo al tratar verdaderamente mucha data con poco o nada de análisis, un rol de esta vertiente sería el Data Arquitect, esa persona encargada de manipular estructurar los datos, manipularlos, masticarlos y dejarlos bien preparados para aquellos encargados de hacer análisis sobre los datos, esta persona trabajaría con Hadoop, Pig, Spark.
Científica: Donde sin que estrictamente se tenga que trabajar con muchísima data (podría ser tanto small Data como Big Data) se lleva a cabo análisis mayormente de tipo estadístico como análisis predictivos, construyendo modelos, un rol de esta vertiente sería la del Data scientist, esa persona encargada de hacer data mining, machine learning, etc.

Y ¿yo qué quiero? ¿qué busco?

YO aspiro a convertirme en un Big Data Architect y seguir aprendiendo y profundizando mis conocimientos en Data mining, Machine learning, estadísticas e investigación de operaciones para forjarme un perfil mas científico.

¿Qué es lo que mas me ha interesado?

Aprendizaje automático (machine learning) y los nuevos enfoques para el procesamiento y almacenamiento de grandes volúmenes de datos (Spark, Flink, Hadoop, Cassandra, HBase).

¿Donde creo que podría aplicarse?

A nivel de ingeniería para almacenamiento por ejemplo de data del hogar (IoT), por ejemplo un termostato inteligente y configurable donde poder gestionar a través de una plataforma de Internet poder gestionar cuando se encienda/apague la calefacción, ver histórico de funcionamiento de la calefacción y sobre estos datos hacer análisis estadísticos.

¿Qué beneficios podría arrojar ?

En base al ejemplo anterior Dinero (ahorro de energía siguiendo con el ejemplo del termostato) o clientes (sistemas de recomendación de productos), otros casos serian por ejemplo tiempo (aprendizaje automático para la elaboración de prestamos preconcebidos o análisis de un cliente al momento de solicitar una hipoteca o préstamo por internet).

Ahora bien, una vez conocido mi concepto de Big Data ¿Coincides con mi definición? ¿discrepas?, ¿Cual es tu concepto del Big Data?