nambrot’s gists

nambrot / spark_trigger.scala

Created December 2, 2017 23:31

	def updateFunction(alertStrings: Seq[String], stateOption: Option[(Boolean, Date, String)]): Option[(Boolean, Date, String)] = {
	(alertStrings, stateOption) match {
	case (_, Some((_, triggerTime, string))) => {
	val cal = Calendar.getInstance()
	cal.add(Calendar.Minute, -60)
	if (triggerTime.before(cal.getTime)) None else Some((false, triggerTime, string))
	}
	case (Nil, None) => None
	case (Seq(string, _*), None) => Some((true, Calendar.getInstance().getTime(), string))
	}

nambrot / spark_log_2.scala

Created December 2, 2017 23:31

nambrot / spark_log.scala

Created December 2, 2017 23:30

nambrot / spark_filter.scala

Created December 2, 2017 23:30

	val alertStream = combinedStream
	.filter {
	case (_, (bloodPressureEvents, heartRateEvents)) =>
	bloodPressureEvents.exists(_.systolic < 100) && heartRateEvents.exists(_.heartRate > 100)
	case _ => false
	}.map {
	case (id, (_, _)) => (id, s"User $id has a problem")
	}

nambrot / spark_join.scala

Created December 2, 2017 23:29

	def reducer(
	a: (List[BloodPressureEvent], List[HeartRateEvent]),
	b: (List[BloodPressureEvent], List[HeartRateEvent])
	) = (a, b) match {
	case ((bp1, hr1), (bp2, h2)) => (bp1 ++ bp2, hr1 ++ h2)
	}

	val combinedStream =
	bloodPressureStream
	.map((event) => (event.userId, List(event)))

nambrot / spark_parse.scala

Created December 2, 2017 23:29

	case class HeartRateEvent(userId: Integer, heartRate: Integer) case class BloodPressureEvent(userId: Integer, systolic: Integer, diastolic: Integer)

	def toEvent[T: Manifest] = (string: String) => { val objectMapper = new ObjectMapper with ScalaObjectMapper objectMapper.setPropertyNamingStrategy(PropertyNamingStrategy.SNAKECASE) objectMapper.configure(DeserializationFeature.FAILONUNKNOWNPROPERTIES, false) objectMapper.registerModule(DefaultScalaModule) objectMapper.readValueT }

	val bloodPressureStream = bloodPressureInput.map(toEvent[BloodPressureEvent]) val heartRateStream = hearRateInput.map(toEvent[HeartRateEvent])

nambrot / spark_setup.scala

Created December 2, 2017 23:28

	val conf = new SparkConf().setMaster("local[2]").setAppName("BiometricAlertStreamProcessor")

	val sc = new SparkContext((conf))
	val ssc = new StreamingContext(sc, Seconds(1))
	val heartRateLines = mutable.Queue[RDD[String]]()
	val hearRateInput = ssc.queueStream(heartRateLines)

	val bloodPressureLines = mutable.Queue[RDD[String]]()
	val bloodPressureInput = ssc.queueStream(bloodPressureLines)

nambrot / gist:6db6304068ddbe3ce6dc

Created April 13, 2015 13:57


	# server.coffee
	renderToString = (route, serializedStoreState, callback) ->
	flux = new Flux()
	flux.deserialize(serializedStoreState)

	Router.run routes, route, (Handler, state) ->
	html = React.renderToString(<FluxComponent flux={flux} render={ => <Handler />}></FluxComponent>)
	embeddedStoreState = "<script>"
	embeddedStoreState += "window.serializedStoreState = " + JSON.stringify(serializedStoreState)

nambrot / gist:bb2b5aae8d1efe16adb7

Created April 13, 2015 13:57


	# app.js
	flux = new Flux()
	flux.deserialize(window.serializedStoreState) if window.serializedStoreState
	Router.run routes, Router.HistoryLocation, (Handler, state) ->
	handler = <FluxComponent flux={flux} render={ => <Handler />}></FluxComponent>
	React.render(handler, document.getElementById("main"))


	# stores/post.coffee

nambrot / c.coffee

Last active August 29, 2015 14:19

Nam Chu Hoai nambrot