IoT-Analyse-Plattform

Internet of Things (IoT) oder auch Industrie 4.0 ist heute in aller Munde. Aber welche Herausforderungen stellen sich eigentlich bei der Verarbeitung großer Datenmengen? Eine Variante kann sein, Daten zu sammeln und später im Batch-Betrieb zu verarbeiten. Meistens ist es jedoch gewünscht, diese ankommenden Daten gleich verarbeiten zu können.

Um mit diesem Mengengerüst an Daten performant umgehen zu können, braucht es eine Plattform, die auf diese Datenmengen reagieren kann. Die Plattform und die dazugehörigen Softwarekomponenten müssen mit der wechselnden Last und vor allem mit den unterschiedlich schnell eingehenden Daten umgehen können. Der SMACK Stack (Spark, Mesos, Akka, Cassandra und Kafka) hat sich als solide Basis für eine solche Plattform erwiesen. Zum einen gibt es die Komponenten, die auf die hereinströmenden Datenmengen reagieren können: Akka, Spark und Kafka. Zum anderen gibt es mit Mesos, Marathon und DC/OS eine Basis-Plattform, die skalieren kann.

Den SMACK Stack in Gänze hat der Kollege Florian Troßbach bereits hier vorgestellt. Dieser Artikel und die dazugehörenden Sourcen fokussieren sich auf den Aufbau der SACK-Komponenten (Spark, Akka, Cassandra and Kafka) des SMACK Stacks. Als Datengrundlage dienen die Bewegungsdaten der Busse der 171-Buslinien von Los Angeles, die dank OpenData frei zur Verfügung gestellt werden und mit einer Rest-API abgefragt werden können. (Metro-API: http://developer.metro.net/)

Mit dem Laden des Videos akzeptieren Sie die Datenschutzerklärung von YouTube.
Mehr erfahren

Video laden

YouTube immer entsperren

Das Video zeigt, wie in Echtzeit die aktuellen Positionsmeldungen der Fahrzeuge in eine “OpenStreetMap”-Karte gezeichnet werden. Gleichzeitig zeigt die Karte die in Cassandra abgelegten vergangenen Positionsdaten. Auf Basis dieser Daten ist auch die Berechnung von Hotspots möglich – Orte, an denen sich besonders viele Fahrzeuge innerhalb eines gewählten Zeitraums befunden haben.

Architekturübersicht

Die Plattform besteht aus mehreren Komponenten:

• Ingest – Akka
• Digest – Spark
• UI – Javascript, Openstreetmap
• Backend – Akka

Der Ingest-Service holt die Informationen von der Metro-API und schreibt initial alle Metadaten direkt in Cassandra. Im Anschluss werden zu allen Routen regelmäßig die aktuellen Bus-Positionsdaten in Kafka geschrieben.
Der Digest liest die Daten aus Kafka, verarbeitet diese mit Spark und schreibt die Positionsdaten aufbereitet sowohl in Cassandra als auch Kafka. In diesem Schritt werden die Positionsdaten für die Abfragen aus dem Frontend optimiert abgelegt.
In einer “OpenStreetMap” Karte werden die so gesammelten Bus-Positionsdaten und Buslinien-Informationen durch die OpenLayers-API angezeigt. Die aktuellen und neuen Positionsdaten werden per WebSocket direkt aus Kafka gelesen und in die Karte gezeichnet.
Die mit einem Spark Job berechneten “Bus-Cluster” können über das Frontend bei Bedarf in der Karte dargestellt werden.
Im Folgenden werden die einzelnen Schritte genauer beschrieben und die Besonderheiten herausgestellt.

Data Ingest

Das folgende Beispiel zum Einlesen von Daten (Ingest) ist exemplarisch für eine IoT-Analyse-Plattform. Da aber nur in einem Intervall von 30 Sekunden die Daten von der öffentlich zugänglichen Stelle abgeholt werden, kann es nicht wirklich mit den Datenmengen und dem damit möglichen Datendruck (back pressure ) verglichen werden, der durch kontinuierlich hereinfließende Daten erzeugt werden kann. Allerdings sind im Vergleich zu vielen anderen Beispielen die gewählten Bus-Daten im 30-Sekunden-Intervall immer noch deutlich dynamischer als z.B. statisch hinterlegte Wetterdaten.

Als Technologie kommt das Scala-Aktoren-Framework Akka zum Einsatz. Beim Starten der Anwendung werden initial die statischenRouteninformationen der Buslinien ausgelesen und in Cassandra abgelegt.

Diese Datenbeinhalten die ID der Route und die entsprechenden Bushaltestellen inklusive Geokoordinaten. Zu jeder Route gibt es noch den Namen der Route, wie er am Bus angezeigt wird.

Diese so gesammelten Metadaten ermöglichen es uns, für alle bekannten Routen die Busdaten periodisch auszulesen. Hierzu gibt es einen Aktor, der sich selbst alle 30 Sekunden anstößt.

1val tick = context.system.scheduler.schedule(0 seconds, 30 seconds, self, Tick())
2 
3override def receive: Receive = {
4  case Tick() => {
5    log.info(s"extracting vehicles Infor for route${routeInfo.id}")
6    extractVicles(routeInfo.id)
7  }
8}

In Akka wird das Konzept der “Reactive Streams ” verwendet, um mit dem möglichen erhöhten Datenaufkommen umgehen zu können. Mehr zu Akka kann man im Blog von Heiko Seeberger erfahren.
Im Folgenden ist der Datenfluss (Flow) für jede Route und den somit zu dieser Route gehörenden Fahrzeugen zu sehen:

1Flow[Vehicle].map(elem => {
2  log.info(s"publishing element: ${elem}")
3  new ProducerRecord[Array[Byte], Vehicle]("METRO-Vehicles", elem)
4}).to(producer).runWith(Source.actorPublisher(VehiclesActor.props(routeInfo, httpClient)))

Als Quelle (source) dient der VehiclesActor, der als Publisher alle 30 Sekunden angestoßen wird. In diesem einfachen Datenfluss werden die eingehenden Daten direkt an einen Kafka Producer gesendet.
Im Publisher werden die Daten abgeholt und über die onNext-Methode an den Datenfluss (Flow) geliefert. Sollte der Flow nicht in der Lage sein, die Daten abzunehmen,werden die Daten im internen Puffer gehalten. Dies ist die von Akka empfohlene Vorgehensweise,mit back pressure umzugehen.

1vehicles.items.foreach {
2  vehicle =>
3  {
4    log.debug(vehicle.toString)
5    log.debug("sending vehicle to stream sink")
6    val vehicleToPersist = Vehicle(vehicle.id, Some(currTime.minusSeconds(vehicle.seconds_since_report).withMillisOfSecond(0).toDate), vehicle.latitude, vehicle.longitude, vehicle.heading, Some(routeInfo.id), vehicle.run_id, vehicle.seconds_since_report)
7    log.debug(s"sending Vehicle ${vehicleToPersist}")
8    if (buffer.isEmpty && totalDemand > 0) {
9      log.info(s"Buffer Empty sending vehicle: ${vehicleToPersist}")
10      onNext(vehicleToPersist)
11    } else {
12      log.info(s"Buffering vehicle: ${vehicleToPersist}")
13      buffer :+= vehicleToPersist
14      if (totalDemand > 0) {
15        val (use, keep) = buffer.splitAt(totalDemand.toInt)
16        buffer = keep
17        log.info(s"Demand is greater 0 sending ${use}")
18        use foreach onNext
19      }
20    }
21  }
22}

Data Digestion

Mit Data Digestion wird die eigentliche Verarbeitungsstrecke der Daten bezeichnet. Innerhalb dieser Verarbeitung können fachliche Informationen extrahiert und optimiert werden. In unserem Beispiel ist es notwendig, die Geokoordinaten der aktuellen Position des Busses so aufzubereiten, dass sie auch einfach aus Cassandra gelesen werden kann. Dies ist notwendig, da sich in einer Cassandra-Tabelle der Schlüssel zum Finden der Daten so zusammensetzt, dass der PartitionKey genutzt wird, um einen Hashwert zu berechnen und weitere Spalten genutzt werden können, um ein Ergebnis einzugrenzen. Denn der Hashwert des PartitionKey bestimmt, auf welchen Knoten im Cluster die Daten abgelegt werden (siehe DataModeling von Datastax ).
Details hierzu und der speziellen Ablage in der Cassandra mit QuadKeys folgen weiter unten.
Ebenfalls werden die so aufbereiteten Daten zur direkten Ausleitung in die Oberfläche auch zurück in Kafka geschrieben.

In unserem einfachen Beispiel ist es die Aufgabe von Spark, die Fahrzeugpositionen (Vehicle) zu TiledVehicle-Objekten zu transformieren und in Cassandra und Kafka abzulegen.
Diese TiledVehicles sind notwendig, damit eine geografische Suche auf den Bus-Positionen gemacht werden kann. So wird jedes Fahrzeug mit einer Kachel-ID versehen, die bestimmt, auf welcher Kachel (Details folgen weiter unten) das Fahrzeug sich gerade befindet. Eine Kachel ermöglicht es, über ein Frontend alle Punkte Innerhalb eines umschließenden Rechtecks – Bounding Box – auf der Karte zu finden.

1val tiledVehicle = vehicle.map(vehicle => TiledVehicle(
2  TileCalc.convertLatLongToQuadKey(vehicle.latitude, vehicle.longitude),
3  TileCalc.transformTime(vehicle.time.get),
4  vehicle.id,
5  vehicle.time,
6  vehicle.latitude,
7  vehicle.longitude,
8  vehicle.heading,
9  vehicle.route_id,
10  vehicle.run_id,
11  vehicle.seconds_since_report
12))
13 
14tiledVehicle.cache()
15 
16tiledVehicle.saveToCassandra("streaming", "vehicles_by_tileid")
17 
18tiledVehicle.foreachRDD(rdd => rdd.foreachPartition(f = tiledVehicles => {
19 
20val producer: Producer[String, Array[Byte]] = new KafkaProducer[String, Array[Byte]](producerConf)
21 
22tiledVehicles.foreach { tiledVehicle =>
23  val message = new ProducerRecord[String, Array[Byte]]("tiledVehicles", new  TiledVehicleEncoder().toBytes(tiledVehicle))
24  producer.send(message)
25}
26 
27producer.close()

Gruppierung von geografischen Punkten anhand eines QuadKeys

Um geografische Punkte, Linien oder Flächen in Cassandra zu finden, müssen diese eindeutig zu orten sein. Dies liegt an der Art der Ablage der Datenstrukturen innerhalb von Cassandra. So können Partitionsschlüssel immer nur auf Gleichheit geprüft werden. Das heißt, möchte man eine Koordinate innerhalb eines umschließenden Rechtecks finden, geht das nicht, sobald die Koordinate Teil des Partitionsschlüssels ist. Daher ist es notwendig, alle Punkte innerhalb einer bestimmten Fläche zu gruppieren. Hierzu wird ein so genannter QuadKey verwendet, wie er auch zur Bestimmung von Microsoft-Bing-Map-Kacheln eingesetzt wird. Hierzu wird die Kartendarstellung der Erde in vier Quadranten aufgeteilt, die sich dann ebenfalls wieder in vier Quadranten unterteilt usw.
Eine visuelle Darstellung hiervon kann auf dem bereits genannten Microsoft Blog gefunden werden.
So ist es möglich, jeden Punkt der Erde mittels eines QuadKeys zu ermitteln; hierzu muss der QuadKey nur lang genug sein. Das gleiche Verfahren wird auch für GeoHashes verwendet, nur sind diese leider nicht menschenlesbar und somit auch nicht einfach zu verifizieren.
Im Kontext dieses Beispieles wurde eine Länge von 15 Zeichen als Länge des Quadkeys festgelegt. Damit ist es möglich, unter Zuhilfenahme einer 256 x 256 Pixel großen Fläche eine geografische Fläche mit ca. 1,5 km Kantenlänge aufzuspannen und als Gruppierung für darin befindliche Koordinaten zu verwenden. Mehr Details dazu sind auf der Seite Bing Maps Tile System zu finden.
Unter Zuhilfenahme des QuadKeys ist es jetzt möglich, alle Punkte innerhalb der Kachel, die diese Fläche darstellt, zu selektieren und auszuwerten.

Darstellung der Livedaten

Im Gegensatz zu statischen Wetterdaten eignen sich dynamische Daten wie die Bewegung von Bussen auch für eine dynamische Darstellung. Somit kommen wir zum letzten Teil der Kette, der Visualisierung. Diese ist simpel gehalten. Es ist eine “OpenStreetMap”-Karte, die per OpenLayers v3 gesteuert wird. Folgende Daten können auf der Karte dargestellt werden:

• Aktuelle Positionen von Fahrzeugen
• Positionen der Fahrzeuge für eine auswählbare Vergangenheit (letzten 15 min.)
• Informationen zu Fahrzeugen und Buslinien
• Darstellung der Buslinien anhand von Haltepunkten
• Darstellung der Cluster-Bildung von Fahrzeugen innerhalb einer Bounding Box.

Für die Darstellung der vergangenen Positionen werden die Daten per REST mit Akka Http aus Cassandra für die übergebene Bounding Box geladen. Bounding Box steht hier für das Rechteck, das den aktuell sichtbaren Kartenausschnitt umschließt.
Für das Einzeichnen aktueller Daten werden die aktuellen Fahrzeugdaten direkt von Kafka per WebSocket in die Oberfläche geladen.
Zum Laden der Fahrzeugdaten innerhalb einer Bounding Box wird mit Akka-Http eine einfache Akka-Http-Route erstellt:

1def vehiclesOnBBox = path("vehicles" / "boundingBox") {
2  corsHandler {
3    parameter('bbox.as[String], 'time.as[String] ? "5") { (bbox, time) =>
4      get {
5        marshal {
6          val boundingBox: BoundingBox = toBoundingBox(bbox)
7 
8          val askedVehicles: Future[Future[List[Vehicle]]] = (vehiclesPerBBox ? (boundingBox, time)).mapTo[Future[List[Vehicle]]]
9          askedVehicles.flatMap(future => future)
10 
11        }
12      }
13    }
14  }
15}

Diese Route stellt eine Anfrage an den entsprechenden VehiclePerBBox-Aktor, um die Fahrzeugdaten für die entsprechende Bounding Box zu erhalten.

1override def receive(): Receive = {
2  case (boundingBox: BoundingBox,time: String) => {
3    log.info("received a BBox query")
4    val eventualVehicles = getVehiclesByBBox(boundingBox, time)
5    log.info(s"X: ${eventualVehicles}")
6    sender() ! eventualVehicles
7  }
8  case _ => log.error("Wrong request")
9}

Es werden analog zum Abspeichern der Daten für die Bounding Box alle in dieser Bounding Box enthaltenen Kachel-QuadKeys (TileIDs) berechnet. Mittels dieser TileIDs und einer entsprechenden Zeitbegrenzung werden die Fahrzeugdaten von Cassandra gelesen.
Damit können diese Fahrzeugdaten in der Karte visualisiert werden.

Entsprechend werden die Informationen zu Buslinien und dem Fahrzeugcluster (siehe weiter unten) geholt.
Die Livedaten wiederum kommen direkt per WebSocket-Verbindung auf die Karte. Dabei wird analog zur REST-Anfrage eine Bounding Box per WebSocket gesendet, mit dem Ergebnis, dass für diese Bounding Box alle aktuellen und auch zukünftigen Fahrzeugdaten übermittelt werden. Das WebSocket-Backend wird durch dieselbe Akka-Http-Anwendung bereitgestellt wie die REST-Endpunkte

Im Request Handler wird geprüft, ob die auf diesem Socket geöffnete Verbindung eine WebSocket-Verbindung ist. Sofern dies der Fall ist, wird ein Akka-Flow zum Abarbeiten der Daten gestartet. Dieser Flow verbindet sich mit einem Aktor zur weiteren Verarbeitung.

1val requestHandler: HttpRequest => HttpResponse = {
2  case req@HttpRequest(GET, Uri.Path("/ws/vehicles"), _, _, _) =>
3  req.header[UpgradeToWebSocket] match {
4    case Some(upgrade) => upgrade.handleMessages(Flows.graphFlowWithStats(router))
5    case None => HttpResponse(400, entity = "Not a valid websocket request!")
6  }
7  case _: HttpRequest => HttpResponse(404, entity = "Unknown resource!")
8}

Der Router-Aktor, mit dem die ein- und ausgehenden Daten per WebSocket übertragen werden, dient lediglich zum Routen aller Messages an alle mit diesem Router verknüpften Aktoren.

1class RouterActor extends Actor with ActorLogging {
2  var routees = Set[Routee]()
3 
4  def receive: Receive = {
5    case ar: AddRoutee => {
6      log.info(s"add routee ${ar.routee}")
7      routees = routees + ar.routee
8    }
9    case rr: RemoveRoutee => {
10      log.info(s"remove routee ${rr.routee}")
11      routees = routees - rr.routee
12    }
13    case msg:Any => {
14      routees.foreach(_.send(msg, sender))
15    }
16  }
17}

Dieser Routen-Aktor ist ab Start der Anwendung mit einem Aktor zum Konsumieren der Fahrzeugdaten von Kafka verknüpft. Daher werden über diesen Router alle neuen Meldungen von Kafka an weitere Konsumenten verteilt.

1class TiledVehiclesFromKafkaActor(router: ActorRef) extends Actor with ActorLogging {
2 
3  import scala.concurrent.ExecutionContext.Implicits.global
4  implicit val materializer = ActorMaterializer()
5 
6  //Kafka
7  val consumerSettings = ConsumerSettings(context.system, new ByteArrayDeserializer, new   TiledVehicleFstDeserializer,
8  Set("tiledVehicles"))
9    .withBootstrapServers("localhost:9092")
10    .withGroupId("group1")
11 
12  val source = Consumer.atMostOnceSource(consumerSettings.withClientId("Akka-Client"))
13  source.map(message => message.value).runForeach(vehicle => router ! vehicle)
14 
15  override def receive: Actor.Receive = {
16    case _ => // just ignore any messages
17  }
18}

Sofern jetzt eine Anfrage mit einer Bounding Box per WebSocket auf dem Router ankommt, wird diese Anfrage an den Publisher weitergeleitet. Gleichzeitig wird ein neuer Publisher-Aktor angelegt und bei dem Router registriert.

1def receive: Receive = {
2  case bbox: BoundingBox => {
3    log.info("received BBox changing behavior")
4    tileIds = TileCalc.convertBBoxToTileIDs(bbox)
5    log.info(s"${tileIds.size} tiles are requested")
6    unstashAll()
7    become(streamAndQueueVehicles, discardOld = false)
8  }
9  case msg => stash()
10}

Sobald hier eine Bounding-Box-Anfrage hereinkommt, verändert sich das Verhalten dieses Aktors, denn nun werden alle in der Bounding Box befindlichen Fahrzeuge per WebSocket an die Webanwendung gesendet.

1def streamAndQueueVehicles: Receive = {
2 
3    // receive new stats, add them to the queue, and quickly
4    // exit.
5    case tiledVehicles: TiledVehicle=>
6      // remove the oldest one from the queue and add a new one
7      if (queue.size == MaxBufferSize) queue.dequeue()
8      if (tileIds.contains(tiledVehicles.tileId)) {
9        queue += Vehicle(tiledVehicles.id,tiledVehicles.time, tiledVehicles.latitude, tiledVehicles.longitude, tiledVehicles.heading, tiledVehicles.route_id, tiledVehicles.run_id, tiledVehicles.seconds_since_report)
10 
11        if (!queueUpdated) {
12          queueUpdated = true
13          self ! QueueUpdated
14        }
15      }
16    // we receive this message if there are new items in the
17    // queue. If we have a demand for messages send the requested
18    // demand.
19    case QueueUpdated => deliver()
20 
21    // the connected subscriber request n messages, we don't need
22    // to explicitely check the amount, we use totalDemand property for this
23    case Request(amount) =>
24      deliver()
25 
26    // subscriber stops, so we stop ourselves.
27    case Cancel =>
28      context.stop(self)
29 
30    case stringMsg:String => {
31      if ("close" == stringMsg) {
32      log.info("closing websocket connection")
33      become(receive, discardOld = true)
34      router ! Cancel
35    }
36  }
37}

Wenn keine weiteren Daten mehr an die WebSocket-Verbindung gesendet werden können, wird der Aktor zerstört.

Clustern von Fahrzeugdaten

Eine Analyseplattform sollte nicht nur kontinuierlich Bewegungsdaten darstellen, sondern auch zeigen, wie auf den gesammelten Daten Analysen gemacht werden können. Hierzu wird exemplarisch eine Clusterbildung auf Fahrzeugpositionen errechnet – an welchen Koordinaten trafen sich die meisten Fahrzeuge innerhalb einer Stunde?
Dabei wird der “Density-Based Clustering in Spatial Databases” (DBSCAN)-Gruppierungs -(Clustering)-Algorithmus angewendet. Als Inspiration für diese Gruppierung dient der Blog von Natalino Busa .
Zuerst werden alle Fahrzeugpositionen für einen bestimmten Zeitraum aus Cassandra selektiert:

1val vehiclesPos:Array[Double] = vehiclesRdd
2  .flatMap(vehicle => Seq[(String, (Double,Double))]((s"${vehicle.id}_${vehicle.latitude}_${vehicle.longitude}",(vehicle.latitude, vehicle.longitude))))
3  .reduceByKey((x,y) => x)
4  .map(x => List(x._2._1, x._2._2)).flatMap(identity)
5  .collect()

Dieses Array wird im Anschluss wieder auf RDDs verteilt und in eine dichte Matrix überführt. Diese Matrix enthält alle zu betrachtenden Koordinaten im Format Latitude, Longitude.

1val vehiclePosRdd: RDD[Array[Double]] = sc.parallelize(seqOfVehiclePos)
2 
3val denseMatrixRdd: RDD[DenseMatrix[Double]] = vehiclePosRdd.map(vehiclePosArray => DenseMatrix.create[Double](vehiclePosArray.length / 2, 2, vehiclePosArray))

Auf dieser dichten Matrix kann dann die eigentliche Berechnung gemacht werden:

1val clusterRdd: RDD[GDBSCAN.Cluster[Double]] = denseMatrixRdd.map(dm => dbscan(dm)).flatMap(identity)

Auf dieser Dichte-Matrix der Längengrade und Breitengrade wird der DBSCAN-Algorithmus angewendet. Hierbei müssen mindestens drei Punkte im Radius des Clusters liegen. Cluster werden innerhalb einer Distanz von ca. 50 Metern gebildet.

1def dbscan(v : breeze.linalg.DenseMatrix[Double]):Seq[GDBSCAN.Cluster[Double]] = {
2  log.info(s"calculating cluster for denseMatrix: ${v.data.head}, ${v.data.tail.head}")
3  val gdbscan = new GDBSCAN(
4    DBSCAN.getNeighbours(epsilon = 0.0005, distance = Kmeans.euclideanDistance),
5    DBSCAN.isCorePoint(minPoints = 3)
6  )
7  val clusters = gdbscan.cluster(v)
8  clusters
9}

Das bereinigte Ergebnis kann wieder in Cassandra geschrieben werden. Auch hier bietet es sich an, die Daten ebenfalls in gekachelter Form in Cassandra zu schreiben.

Besonderheiten

Der Ingest und das Backend für die UI wurde komplett mit Akka und Scala 2.11 geschrieben. Da es aber im Kontext von SMACK zu empfehlen ist, Spark mit Scala 2.10 zu nutzen, musste die Anwendung so gebaut werden, dass prinzipiell beide Scala-Versionen unterstützt werden. Hierzu wird das SBT-Plugin sbt-doge verwendet, das einen Cross-compile im Build-Prozess unterstützt. Dies ist besonders sinnvoll für die gemeinsam genutzten Case-Klassen und die im Commons-Modul befindliche Hilfsklasse zum Berechnen von Kachel-IDs. Auch die Serialisierer und Deserialisierer für die Case-Klassen im Kontext von Kafka befinden sich im Commons-Modul.
Als Serialisierer wird die Fast-Serialization verwendet. Diese ermöglicht sogar, Case-Classes mit Scala 2.11 zu serialisieren und diese mit Scala 2.10 zu deserialisieren.
Beim Erstellen des Multi-Modul-Projektes wurde darauf geachtet, dieses so einfach und übersichtlich wie möglich zu halten. Daher befinden sich alle Konfigurationen bzgl. Build auch nur in der Wurzel-build.sbt-Datei.

Zusammenfassung

Die hier gezeigte Analyseplattform lässt sich sicherlich nicht nur für IoT-Daten verwenden, sondern auch für andere große Aufkommen von Datenmengen einsetzen. Besonders bei großen Datenmengen ist solch eine Verarbeitungslinie besser in der Lage, mit dem anfallenden Mengengerüst umzugehen, als z.B. eine REST-Anwendung, die die Daten in eine RDBMS ablegt.
Neben den vielen Vorteilen einer Cassandra-Datenbank als Speicherungsmedium zeigt sich aber auch der mögliche Nachteil einer solchen Lösung. In einer RDBMS-Datenbank ist es möglich, Größer- und Kleiner-Vergleiche auf dem primären Schlüssel abzufragen. Dies unterstützt Cassandra so nicht. Mithilfe der Gruppierung der Datenpunkte per QuadKey lässt sich dieses Problem allerdings addressieren und Cassandra kann weiterhin mit performaten Antwortzeiten punkten.
Der Punkt der Kachel-IDs (QuadKeys) lässt sich ggf. auch noch weiter optimieren. Die hier gezeigte Lösung setzt zur Zeit auf ein statisches Zoomlevel auf; es wäre aber durchaus sinnvoll und machbar, die Kacheln und dazugehörigen QuadKeys für weitere Zoomstufen zu berechnen und somit größere Bereiche zusammenzufassen. Eine Abfragestruktur müsste mit diesen verteilten Kacheln umgehen können. Hierzu bieten sich Hilbert Space-Filling bzw. Z-Kurven an.

LINKS:

• https://github.com/ANierbeck/BusFloatingData
• http://developer.metro.net/
• https://ruedigermoeller.github.io/fast-serialization/
• https://github.com/sbt/sbt-doge
• https://blog.codecentric.de/2016/05/der-smack-stack-zum-anfassen/
• https://msdn.microsoft.com/en-us/library/bb259689.aspx
• https://www.oreilly.com/ideas/clustering-geolocated-data-using-spark-and-dbscan
• https://www.oreilly.com/people/b7f1d-natalino-busa
• http://www.reactive-streams.org/
• https://blog.codecentric.de/en/2015/07/a-map-of-akka/
• https://people.csail.mit.edu/jaffer/Geometry/HSFC
• http://blog.notdot.net/2009/11/Damn-Cool-Algorithms-Spatial-indexing-with-Quadtrees-and-Hilbert-Curves
• https://www.datastax.com/dev/blog/basic-rules-of-cassandra-data-modeling