Streaming Analytics with Apache Pulsar and Spark Structured Streaming

Producer: Pulsar

We’ll simulate these events as we described in the Pulsar introduction article. First, we’ll set up a Docker container that will run Pulsar:

docker run -it \
  -p 6650:6650 \
  -p 8080:8080 \
  --name pulsar \
  apachepulsar/pulsar:2.8.0 \
  bin/pulsar standalone

And in another terminal we’ll create a topic events to accept our values:

docker exec -it pulsar bash
bin/pulsar-admin topics create events

Now, in our Scala application, we need to set up a Pulsar producer. Assuming we can parse our data easily (in this case from a file with randomized, realistic data), we’ll need to set up the necessary Pulsar scaffolding:

import com.sksamuel.pulsar4s.{DefaultProducerMessage, EventTime, MessageId, ProducerConfig, PulsarClient, Topic}
import com.sksamuel.pulsar4s.circe._
import io.circe.generic.auto._
import io.circe.{Decoder, Encoder, Json, Printer}
import org.apache.pulsar.client.api.Schema
import org.apache.pulsar.common.schema.{SchemaInfo, SchemaType}

lazy val topicName      = "events"
lazy val eventsFilePath = "src/main/resources/events.csv"
lazy val serviceUrl     = "pulsar://localhost:6650"
lazy val adminUrl       = "http://localhost:8080"
lazy val producerName   = "event-producer"
lazy val threadPoolSize = 4

// events parsed elsewhere
val events: List[Event] = loadEvents(eventsFilePath, withHeader = true)

val pulsarClient = PulsarClient(serviceUrl)
val topic = Topic(topicName)

val eventProducer = pulsarClient.producer[Event](
  ProducerConfig(
    topic,
    producerName = Some(producerName),
    enableBatching = Some(true),
    blockIfQueueFull = Some(true)
  )
)

Then we need to process all events in the list by sending them asynchronously to Pulsar. The following is a quick sketch of how it would look like:

val messageIdFutures: Seq[Future[MessageId]] = events.map { event =>
  Thread.sleep(10)

  // marshal an event for Pulsar
  val message: DefaultProducerMessage[Event] = DefaultProducerMessage[Event](
    Some(event.userid),
    event,
    eventTime = Some(EventTime(event.eventTime)))

  //send it
  val messageIdFuture = eventProducer.sendAsync(message)

  // being quite verbose to track messages
  messageIdFuture.onComplete {
    case Success(id) => println(s"Ack received for message with id '$id'.")
    case Failure(exception) => println(s"Failed to sent message - Reason: $exception")
  }

  // keep the Future for analysis at the source, e.g. check message IDs
  messageIdFuture
}

We can then use the resulting Futures to track message IDs and whether they’ve been sent successfully or not, etc. For this demo, we’re just going to close the application once it’s done.

Future.sequence(messageIdFutures) // turn the Seq[Future[...]] into Future[Seq[...]]
  .andThen {
    case Success(_) => println("Producer finished sending event records.")
    case Failure(e) => println(s"A failure occurred. Check the logs and the stack trace: $e")
  }
  .andThen {
    case _ => eventProducer.close()
  }

Consumer: Spark Structured Streaming

Spark Structured Streaming — which we teach here at Rock the JVM — is a stream computing engine provides more advanced features that are helpful to our use case:

• Support for Session Windows - we can create session windows based on the eventTime and the userSession id.
• Support for groupBy aggregations - we want to aggregate the eventTypes on the user session windows.
• Watermarking support - allows for late event handling.
• Filtering on complex data types - filter rows based on a list filtering predicate.

With our dependencies in place, let’s first connect to Pulsar and start reading events from our events topic.

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.OutputMode

val clickEventsDF = spark
        .readStream
        .format("pulsar")
        .option("service.url", "pulsar://localhost:6650")
        .option("admin.url", "http://localhost:8080")
        .option("topic", "events")
        // the following two configs are optional for this demo but important if you're doing authenticated data access
        .option("pulsar.client.authPluginClassName","org.apache.pulsar.client.impl.auth.AuthenticationToken")
        .option("pulsar.client.authParams","token:you-token<>")
        .load()

We can easily connect to Pulsar with Spark with the connection URLs, input topic name, and preferred security method if security is enabled. You can check the full list of the available configuration options at the Pulsar connector’s GitHub..

With the connection in place, we’ll transform the data to a custom data type useful for analysis:

case class AnalysisEvent(userSession: String, userId: String, eventTime: Timestamp, eventType: String)

import spark.implicits._

val parsedEvents = clickEventsDF
  .selectExpr("CAST(value AS STRING)").as[String]
  .map { line =>
    val event = io.circe.parser.decode[Event](line).getOrElse(Event.empty())
    AnalysisEvent(event.userSession, event.userid, new Timestamp(event.eventTime), event.eventType)
  }.as[AnalysisEvent]

And with this Dataset in place, we can now sketch an example implementation:

• Discarding events older than 30 minutes
• Using session windows of 80 minutes, now automatic in Spark 3.2
• Aggregating event types
• Looking for shopping cart events

val checkoutEvents = parsedEvents
  .withWatermark("eventTime", "30 minutes")
  .groupBy(col("userSession"), col("userId"),  session_window(col("eventTime"), "80 minutes"))
  .agg(collect_list("eventType").as("eventTypes"))
  .filter(array_contains(col("eventTypes"),"cart"))
  .select(
    col("session_window.start").as("sessionWindowStart"),
    col("session_window.end").as("sessionWindowEnd"),
    col("userId"),
    col("userSession"),
    col("eventTypes")
  )

The output should look something like this:

+-------------------+-------------------+---------+------------------------------------+------------------------------------------------------------------------------------------------------------------------------+
|sessionWindowStart |sessionWindowEnd   |userId   |userSession                         |eventTypes                                                                                                                    |
+-------------------+-------------------+---------+------------------------------------+------------------------------------------------------------------------------------------------------------------------------+
|2019-11-01 02:21:03|2019-11-01 02:56:28|564133858|ef95cac1-6580-44f0-a10b-c7e836982154|[view, view, cart, cart, purchase, view, view, view, cart, view]                                                    |
|2019-11-01 00:59:55|2019-11-01 01:34:27|566286947|b86c0313-d0cd-4222-888f-2ed03a1d697f|[view, view, view, cart, purchase, view]                                                                                      |
|2019-11-01 02:22:39|2019-11-01 02:54:45|528271890|b425dd2d-dae4-4915-8e8f-b1d4d2717471|[view, cart, view, view]                                                                                                      |
|2019-11-01 01:49:20|2019-11-01 02:21:31|553414803|5006939e-fbe7-4e1c-809c-ffa3b16eb20c|[view, view, view, cart, view]                                                                                                |
|2019-11-01 01:54:59|2019-11-01 02:29:24|556325945|ad789442-8756-41d7-b05a-11b90124032d|[view, view, view, cart, purchase, view, cart, cart, cart, cart, cart, cart, cart, cart, cart, cart, cart]                    |
|2019-11-01 00:37:42|2019-11-01 01:14:30|549630515|8847ab0c-1f0b-42fb-9ff4-f44b9f523b4b|[view, view, view, view, view, cart, view, view, view, view, cart, view, view, view]                                          |
|2019-11-01 00:13:32|2019-11-01 00:44:49|563558500|e0729b6c-eafe-4b0f-9d66-6ee777d08488|[view, cart, view, view, view, view, view]                                                                                    |
|2019-11-01 02:15:30|2019-11-01 02:49:24|512411198|64a9195f-e4ee-448f-9241-9b4d23467f5d|[view, cart, view, view, cart, view]                                                                                          |
|2019-11-01 01:44:05|2019-11-01 02:15:30|515742799|ca33e1b9-ecf5-4b50-ba76-007248bad43d|[view, cart, cart, cart, cart, view]                                                                                          |
|2019-11-01 00:41:39|2019-11-01 01:12:36|557332447|70c0ccdf-9452-49cc-bfa5-568096d64680|[view, cart, view]                                                                                                            |
|2019-11-01 01:40:47|2019-11-01 02:34:50|520761071|6dad05da-a718-4f05-92bc-1ed9796404cd|[view, view, view, view, view, view, view, view, view, view, view, cart, purchase]                                            |
|2019-11-01 00:37:04|2019-11-01 01:08:52|562200921|ca5a71f1-33c8-4fcd-b793-5fcea6f260c0|[view, cart, purchase, view]                                                                                                  |
|2019-11-01 00:10:07|2019-11-01 00:46:10|516426931|ef4867dd-b922-4d92-abec-9a75acb2b769|[view, view, view, cart, purchase, view, cart, cart, view, view]                                                    |
|2019-11-01 02:07:33|2019-11-01 02:45:30|554459781|c43b43dd-dc54-4d1c-bfd8-3bcbdfb94870|[view, cart, purchase, view, view]                                                                                            |
|2019-11-01 02:11:34|2019-11-01 02:52:13|539100846|365b1088-8a4f-47e4-a6e9-e66d56b1b998|[view, cart, cart, cart, view, cart, view, view, view, view, view]                                                            |
|2019-11-01 01:34:14|2019-11-01 02:08:51|519456951|f2136bd5-a50d-4e05-8a90-27b288065459|[view, cart, cart, purchase]                                                                                                  |
|2019-11-01 01:47:49|2019-11-01 02:26:31|515561413|98a12974-b589-48fe-b6f5-a55cd844cdd8|[view, view, view, view, cart, view]                                                                                          |
|2019-11-01 01:29:06|2019-11-01 02:00:39|518913698|afa1ad69-55bb-4ef8-9d02-6648596ca5ec|[view, cart, purchase, view]                                                                                                  |
|2019-11-01 01:53:53|2019-11-01 02:36:26|556601011|45658f52-9e11-45fa-a8d6-9414d349aa4d|[view, view, view, view, view, view, cart, cart, view, view, view, view, view, view, view, view, view, view, view, view, view]|
|2019-11-01 01:01:28|2019-11-01 01:34:04|538178630|9cf447dc-7aa8-47f2-8c3b-44318cf5608a|[view, view, view, cart, cart, cart, purchase, view]                                                                          |
+-------------------+-------------------+---------+------------------------------------+------------------------------------------------------------------------------------------------------------------------------+

Some notes on session windows:

• A Session window sizes dynamically based on the window length, depending on the inputs.
• A Session window starts with an input, and expands itself if the following input is received within the gap duration.
• For a static gap duration, a session window closes when there’s no input received within gap duration after receiving the latest input.

By using session windows and grouping on the user sessions, we can easily see if a user has a cart event type, but no purchase event. This makes it easy to filter the events we consider as abandoned cart and forward them to a downstream topic for further processing. An example action could be to send out a reminder notification (via smartphone app if available, or via email) to the users after some period of time, e.g. an hour.

Parsing Events

This bit should not be the front and center of the article as the focus is on Pulsar and Spark, so for your convenience, this is the simple code we used to parse the events in this csv file.

import java.sql.Timestamp
import java.util.concurrent.{ExecutorService, Executors}
import scala.concurrent.{ExecutionContext, Future}
import scala.io.BufferedSource
import scala.util.{Failure, Success, Try}

def loadEvents(path: String, withHeader: Boolean = false): List[Event] = {
  val r = Try(scala.io.Source.fromFile(path))
    .map(source => sourceHandler(source, withHeader))
    .fold(_ => List.empty, identity)

  val result = Try(scala.io.Source.fromFile(path)) match {
    case Success(source) =>
      Some(sourceHandler(source, withHeader))
    case Failure(exception) =>
      println(s"Failed to read file '$path' - Reason: $exception")
      None
  }

  result.getOrElse(List.empty)
}

private def sourceHandler(source: BufferedSource, withHeader: Boolean): List[Event] = {
  val events: List[Event] = source.getLines()
    .map(toEvent)
    .toList

  if (withHeader) events.drop(1) else events
}

private def toEvent(line: String): Event = {
  Try {
    val tokens = line.split(",")
    val eventTime = Timestamp.valueOf(tokens(0).replace(" UTC", ""))

    Event(
      tokens(7),
      eventTime.getTime,
      tokens(1),
      tokens(2),
      tokens(3),
      tokens(4),
      tokens(5),
      tokens(6).toDouble,
      tokens(8)
    )
  }.getOrElse(Event.empty())
}