migue

@Component(
	immediate = true,
	service = ModelListener.class
)
public class RatingsEntryModelListener extends BaseModelListener<RatingsEntry> {

	@Override
	public void onAfterCreate(RatingsEntry model)
		throws ModelListenerException {

migue

@Component(
	immediate = true,
	service = ModelListener.class
)
public class MBMessagesModelListener extends BaseModelListener<MBMessage> {

	@Override
	public void onAfterCreate(MBMessage model)
		throws ModelListenerException {

migue

val sparkConf = new SparkConf().setAppName("KafkaMessageBoardsCollector")

    val ssc =  new StreamingContext(sparkConf, Seconds(5))

    ssc.checkpoint("checkpoint")

    val topicMap = topics.split(",").map((_,numThreads.toInt)).toMap

    val kafkaStream = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)

migue

val messagesTable = sc.textFile(messagesInput).map(_.split(";")).map(m => MBMessage(m(4), m(5)))

val schema =
StructType(
   StructField("title", StringType, false) ::
   StructField("body", StringType, true) :: Nil)

val messagesSchemaRDD = sqlContext.applySchema(messagesTable, schema)

messagesSchemaRDD.registerTempTable("messagesTable")

migue

package com.liferay.message.boards.classifier

import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.mllib.feature.HashingTF

object Utils {

  val numFeatures = 1000
  val tf = new HashingTF(numFeatures)

migue

val ssc =  new StreamingContext(conf, Seconds(5))

val topicMap = topics.split(",").map((_,numThreads.toInt)).toMap

val kafkaStream = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)

val model = new KMeansModel(ssc.sparkContext.objectFile[Vector](modelFile.toString).collect())

val filteredMessages = kafkaStream
  .filter(m => model.predict(Utils.featurize(m)) == clusterNumber.toInt)

migue

import scala.util.Random

def winner(members:List[String]):String = {
  new Random(System.currentTimeMillis()).shuffle(members).head
}

// write the names of all the members involved in the raffle

val qcon = List("Jorge Baez", "Daniel Jiménez", "Gonzalo Gómez", "Jorge Nota", "Jose Ignacio Dominguez", "Daniel Hernández", "Manuel Carrasco")

migue

/**
 * Domain model representing the metrics of our system. All the metrics at our
 * system should implement this interface
 * 
 * @author migue
 * 
 */
public interface Metric {

	public String name();

migue

...

void activate(ComponentContext componentContext) {

System.out.println("Starting the metrics manager. Waiting for metrics providers . . .");

_executorService.execute(new Runnable() {

	@Override
	public void run() {

migue

<scr:component xmlns:scr="http://www.osgi.org/xmlns/scr/v1.1.0" name="metrics" immediate="true">

  <scr:implementation class="com.github.migue.metrics.internal.manager.MetricsManager" />

	<scr:reference
		interface="com.github.migue.metrics.spi.MetricsProvider"
		cardinality="0..n"
		policy="dynamic"
		bind="addMetricsProvider"
		unbind="removeMetricsProvider"/>