sparkstreaming.scala

val testData = 
  ssc.textFileStream(...).map(LabeledPoint.parse) // Infinite stream

testData.predict { case testDf =>
  // Explode features in input streams
  val testDf_exploded = ...
  testDf_exploded
    .join(model, testDf_exploded("feature") === model("feature"), "LEFT_OUTER")
    .select($"rowid", ($"weight" * $"value").as("value"))
    .groupby("rowid").sum("value")
    .select($"rowid", sigmoid($"SUM(value)"))
}