# Streaming data

Source: https://orbitalhq.com/docs/guides/streaming-data

**Warning: We're working on it...**
This guide is under construction. If you hit some rough edges, or find it confusing, come yell at us [on slack](https://join.slack.com/t/orbitalapi/shared_invite/zt-697laanr-DHGXXak5slqsY9DqwrkzHg).

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

## Describing the data sources

_Image: 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:

```taxi taxonomy.taxi
namespace petflix

type FilmId inherits String
type FilmTitle inherits String
type ReviewText inherits String
type PerformanceDate inherits DateTime
type TheatreName inherits String
```

### Adding the Kafka broker

Configure the Kafka connection in `connections.conf`:

```hocon 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:

```protobuf
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:

```taxi 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:

```taxi
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](https://developer.confluent.io/faq/apache-kafka/kafka-clients/#kafka-clients-what-is-groupid-in-kafka)

```hocon 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:

```taxi
@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`