Project Fortis\u00a0is a social data ingestion, analysis, and visualization platform. Originally developed by Microsoft in collaboration with the\u00a0United Nations Office for the Coordination of Humanitarian Affairs<\/a>\u00a0(UN OCHA), Fortis provides planners and scientists with tools to gain insight from social media, public websites and custom data sources. We’ve explored previous work on Project Fortis in other code stories on this blog, such as\u00a0Project Fortis: Accelerating UN Humanitarian Aid Planning with GraphQL<\/a>\u00a0and\u00a0Building a Custom Spark Connector for Near Real-Time Speech-to-Text Transcription<\/a>.<\/p>\n In Project Fortis, we use Spark Streaming to consume and process real-time news and events from a variety of sources, including social media outlets such as Twitter and Facebook. This article assumes familiarity with Apache\u00a0Spark Streaming application development<\/a>.\u00a0\u00a0<\/strong>As a part of this processing, we perform sentiment and geolocation analysis as well as entity extraction, ultimately aggregating this information into a Cassandra database, which powers a web service used to visualize it.<\/p>\n One notable project requirement for Fortis is that many aspects of the processing done by Spark must be user-tunable; for example, our Spark application makes use of both a whitelist and a blacklist to filter out incoming events that are uninteresting to the user. These lists, among other settings, are maintained through an admin dashboard from within our web service.\u00a0 It’s important that the user is able to update these lists at any time.<\/p>\n To satisfy this requirement of our work on Project Fortis, we needed to be able to communicate with Spark to reconfigure our data processing logic at runtime in response to configuration changes from an external service. In this code story, we’ll explore how we were able to achieve this goal using Spark transformations and Azure Service Bus.<\/p>\n <\/p>\n One of Spark’s limitations is that it’s not possible to change an application’s\u00a0 For example,\u00a0 While we cannot change pipeline stages or their ordering, we can use generic transformations and output actions that operate at the\u00a0 Functions (callbacks) we provide to For a deeper understanding of the approach covered here, check out our supplemental blog post, Spark Streaming Transformations\u00a0: A Deep-dive<\/a>, which captures some non-obvious behaviors and limitations of Spark which we’d uncovered throughout its design.<\/p>\n By writing some setup code to run on the driver for each incoming For Project Fortis’s requirements, it made the most sense to place the code to do this within a When using\u00a0 However, this approach itself added additional complexity due to serialization concerns around Spark checkpointing. We needed to find a way to share the mutable state between the Spark thread executing our We found the most elegant solution to the above was to declare the mutable state through which these threads would communicate as static memory. In this way, the Spark thread(s) always refer to the current copy, as does startup code. This works because static fields are not serialized, and instead follow the lifetime of the process; our When designing Project Fortis, we chose to use Azure Service Bus as the external service from which we’d receive runtime configuration changes. Azure Service Bus is a highly available reliable message queue service which can be used to achieve exactly-once semantics for messages, which is what we wanted for our use cases. Furthermore, its API supports long-polling, allowing our Spark application to listen on a socket under the hood in order to begin processing configuration changes immediately.<\/p>\n To best illustrate the concepts of the approach we took with Project Fortis, we’ve created\u00a0a bare-bones sample project<\/a>. In the sections below, we’ll explore its design. While the sample is based on Azure Service Bus, the approach it illustrates can be easily extended for use with other external services.<\/p>\nIntroduction<\/h2>\n
Runtime Reconfiguration<\/h2>\n
DStream<\/code>\u00a0graph after the streaming context has been started. From an API perspective, this graph is comprised of all DStream<\/code> transformation lineages which lead from an output operation (i.e.,print<\/code>foreachRDD<\/code>, etc.) back to a source input stream (receiver, direct input stream, etc.). In other words, the hierarchy of the\u00a0DStream<\/code>\u00a0graph defines the configuration and ordering of the stages which comprise an application’s data-processing pipeline(s).<\/p>\nmyDStream.transform(...).map(...).filter(...).print()<\/code>defines a pipeline in which all RDD<\/code>s arriving on the stream “myDStream<\/code>” will be transformed by some function, mapped by another, filtered by yet another, and finally printed.<\/p>\nRDD<\/code>-level\u00a0 (namely transform<\/code>\u00a0and foreachRDD<\/code>, respectively) which give us a chance to run arbitrary code for each incoming RDD<\/code>\u00a0(batch).<\/p>\nforeachRDD<\/code>\u00a0and transform<\/code>\u00a0are executed\u00a0within the driver process. They’re designed to build an execution plan for each RDD<\/code>; they set up the\u00a0RDD<\/code>-level transformations that will occur for each individual\u00a0RDD<\/code>, and scheduling is done from the Spark driver. We were able to make use of this to satisfy our goal; before we process each RDD<\/code>, we poll an external service for configuration changes and adjust the logic applied to the current and future\u00a0RDD<\/code>s.<\/p>\nDesigning our Approach<\/h2>\n
RDD<\/code>, we can reference an external service and reactively change how we process data (i.e., in response to another service). Useful external services for this purpose include databases, web services, and messaging queue services like Azure Service Bus (more on this option below), but the options are of course limitless.<\/p>\ntransform<\/code>\u00a0callback (a transformation). We chose this over foreachRDD<\/code>\u00a0(an output operation) since it allowed us to modify our configuration the farthest upstream within our data pipeline. We were then able to use the resulting DStream<\/code>\u00a0to supply data to completely separate downstream transformation sequences (data pipelines). For simpler applications with a single output operation, using foreachRDD<\/code>\u00a0should suffice.<\/p>\ntransform<\/code>, however, work done in the callback must not be long-running (a Spark thread will invoke the callback). This limitation rules out network communication, making it more difficult to communicate with external services. To solve this, we used a worker\/background thread to fetch and prepare data from our external service, passing it off to the Spark driver thread(s) only once ready.<\/p>\ntransform<\/code>\u00a0callback and our background thread, which we’d started prior to the streaming context, near our application’s entry point. This problem is somewhat non-trivial given that the state of the transform<\/code>\u00a0function is serialized by Spark during checkpointing, so upon recovery\/deserialization, no direct references from this code to or from objects we had control over in the old JVM process would exist.<\/p>\ntransform<\/code>\u00a0callback would have a code reference to the shared state. The implication of this is that we needed to ensure that the state was initialized properly prior to starting the Spark streaming context, even after recovery from checkpointing. In practice, we could just add some more initialization code again at the entry point to our application both before running the background thread and (transitively) before we telling the streaming context to start.<\/p>\nAzure Service Bus<\/h2>\n
Sample Project<\/h3>\n