| func SetupProfiler() (*http.Server, chan error, error) { | |
| profilerRouter := chi.NewRouter() | |
| profilerRouter.Mount("/profile", Profiler()) | |
| server := http.Server{ | |
| Addr: fmt.Sprintf(":%v", 8080), | |
| Handler: profilerRouter, | |
| } | |
| errChannel := make(chan error) |
A short course to introduce my students to the Go programming language.
I try to provide my students with a comprehensive, and idiomatic view of the Go programming language, focusing on different topics ranging from language syntax, design, and good practices to concurrency, testing or profiling.
This chapter will try to cover the basic concepts and syntax of the language:
I hereby claim:
To claim this, I am signing this object:
| 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) |
| 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) |
| 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") |
| 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) |
| @Component( | |
| immediate = true, | |
| service = ModelListener.class | |
| ) | |
| public class MBMessagesModelListener extends BaseModelListener<MBMessage> { | |
| @Override | |
| public void onAfterCreate(MBMessage model) | |
| throws ModelListenerException { |
| @Component( | |
| immediate = true, | |
| service = ModelListener.class | |
| ) | |
| public class RatingsEntryModelListener extends BaseModelListener<RatingsEntry> { | |
| @Override | |
| public void onAfterCreate(RatingsEntry model) | |
| throws ModelListenerException { |