Orbital

Guides

Streaming data

This guide shows how to build bespoke, enriched event streams from Kafka.

Describing the data sources

The architecture for this demo

The architecture for this demo

The architecture for this demo

Our demo has a few services running, which we’ll join together to create a bespoke event stream:

  • Kafka Service: A Topic that publishes a message whenever a new review is published
  • FilmsDb: A database containing a list of films
  • ListingsService: A REST API that tells us where our movies are currently playing

Our services share a taxonomy used to describe the elements we can fetch:

taxonomy.taxi
namespace petflix
type FilmId inherits Stringtype FilmTitle inherits Stringtype ReviewText inherits Stringtype PerformanceDate inherits DateTimetype TheatreName inherits String

Adding the Kafka broker

Configure the Kafka connection in connections.conf:

connections.conf
kafka {    "myKafkaBroker" {        connectionName=myKafkaBroker        connectionParameters {            brokerAddress="localhost:9092"            groupId=orbital        }    }}

Describing the Kafka topic

Next, add some Taxi metadata to the message written onto our Kafka topic:

syntax = "proto3";import "taxi/dataType.proto";
message NewReviewPostedMessage {  int32 filmId = 1 [(taxi.dataType)="petflix.FilmId"];  string reviewText = 2 [(taxi.dataType)="petflix.ReviewText"];}

Finally, declare the Kafka topic:

reviews.taxi
service KafkaService {  stream reviews : Stream<NewReviewPostedMessage>}

Creating an event stream

We can create a dedicated event stream, with the data we need by submitting the following query to Orbital:

stream { NewReviewPostedMessage } as {  // We define the field names that matter to us.  // Orbital matches on data types, as field names often differ between systems.   id : FilmId   review : ReviewText
  name : FilmTitle // Fetched by a database call  nextPerformances: FilmListing[] // Fetched by a REST API call}

Specifying The GroupId For Kafka Subscriptions

When Orbital subscribes to a Kafka Topic for a streaming query, it will use groupId connection paremeter defined in connections.conf to set the value of group id for the corresponding kafka consumer

connections.conf
kafka {    "myKafkaBroker" {        connectionName=myKafkaBroker        connectionParameters {            brokerAddress="localhost:9092"            groupId=orbital        }    }}

@ In the above configuration, the groupId is set to orbital. However, you can override groupId value in your queries by using the @StreamConsumer annotation. Here is an example:

@StreamConsumer(id = "reviewStreamGroup")stream { NewReviewPostedMessage } as {  id : FilmId   review : ReviewText
  name : FilmTitle // Fetched by a database call  nextPerformances: FilmListing[] // Fetched by a REST API call}

The Kafka Consumer created for the above streaming query will set the group id to reviewStreamGroup