First steps with Streammachine.io
Published: 2020-10-28 13:56:28


This post aims to show some *first steps* with the Stream Machine platform. The target audience is developers, data engineers and possibly data scientists. This post uses Python, but it can similarly be done with Stream Machine's [Other language drivers](https://mvnrepository.com/search?q=%20streammachine).

**About me:** I'm Bart van Deenen, lead engineer of Stream Machine, so I should sort of know what I'm talking about :-). This post is meant to be unbiased, and from the perspective of a third-party developer.

## Stream Machine

[Stream Machine](https://portal.streammachine.io/) promises to provide `Lightning fast, privacy secured, customer data - you can actually use`.

So what does this actually mean?

* Stream Machine accepts events with a strictly defined serialization schema
  (currently [Apache Avro](https://avro.apache.org/) and
  [Json-Schema](https://json-schema.org/) are supported). Any valid schema can
  be used, but needs to be registered with Stream Machine.
* Stream Machine handles events that are subject to an event-contract that
  defines which fields in the event schema (the _serialization_ schema) contain
  _Personally Identifiable Information_ ( referred to as _PII_ or _PII Data_).
* Stream Machine event contracts may contain customizable validation rules that
  define field value validity.
* Stream Machine processes events with a highly available fault tolerant stream
  processing system that _encrypts all PII field values_. The encryption keys are
  rotated every 24 hours, and this leads to a GDPR compliant stream of event
  data that can be used by everyone in your company. During the 24 hours, the
  encrypted values remain static.
* Stream Machine events contains *consent-level* information, and only those events that
  allow decryption of PII data for certain purposes can be decrypted into
  *decrypted stream(s)* that can only be used by those inside your company that
  are allowed.

This post uses a debugging output of the stream data that uses a websocket.  Production level output streams require hooking up our internal [Apache Kafka](https://kafka.apache.org/) streams. AWS S3 and Google Cloud Storage buckets can be used for batch processing. This will be explored in a next blog post.

## The plan

I'm going to build a Python application that mimics users clicking around on a dummy web-shop, that will send a click stream to Stream Machine. I want to retrieve the anonymized data from a Google Cloud bucket, and show them in a Jupyter notebook. I also want to see that only for those simulated users that have given full *personalized marketing* permissions I retrieve their click stream events. This first post just gets the basics working, i.e. sending events to Stream Machine, and retrieving them.

## The steps

### An account

I went to [streammachine.io](https://portal.streammachine.io/) to register an account, and after confirming my email, I was shown this page:

![login](https://imgur.com/s2qZwhe.png)

### Get the CLI
The [cli](https://github.com/streammachineio/cli) is the most mature way to interact with Stream Machine. Install the most recent version and don't forget to install the shell auto-completion.

## Let's create a new stream

```
strm auth login your-email@wherever
strm create stream demo --save
```
The `--save` options causes the credentials to be saved in
`~/.config/stream-machine/Stream/demo.json`.

## Let's send an event!

I'm following along with the [Python example in the documentation](https://github.com/streammachineio/python-examples). I'm going to use the [syncsender](https://github.com/streammachineio/python-examples/blob/master/examples/syncsender.py) because I want to play with it in _ipython_.

The demo example uses the `streammachine/demo/1.0.2` schema. You can see its full definition via `strm get schema streammachine/demo/1.0.2`. The corresponding event-contract that defines validation rules and such is `streammachine/example/1.3.0`.

For convenience I've copied the saved credentials file (see above) into the working directory.


### Keep an eye on the stream

    strm egress demo

This starts a websocket connection where you'll see a json serialization of the messages on a stream.

```
python3 -m venv .venv
. .venv/bin/activate
make
# observe the installed streammachine stuff.
pip freeze | grep streammachine
streammachine-driver==1.0.0
streammachine-schemas-common==1.0.0
streammachine-schemas-demo-avro==1.0.2
```

Ok, let's start _ipython_

```
python3 -m pip install ipython
$> ipython

# import a class that matches the structure of the demo schema
# note that this package name is so long because it's derived from the Stream
# Machine schema ref streammachine/demo and also the `namespace` inside the Avro
# schema (which is io.streammachine.schemas.demo.v1)

from streammachine_io_streammachine_schemas_demo_v1.io.streammachine.schemas.demo.v1 import DemoEvent

from streammachine.driver import StreamMachineEvent, current_time_millis
from streammachine.driver.client.syncsender import SyncSender

event = DemoEvent()

import json
creds = json.load(open("demo.json"))
sender = SyncSender(creds['ref']['billingId'], creds['credentials'][0]['clientId'],
    creds['credentials'][0]['clientSecret'])
sender.start()
sender.wait_ready()
sender.send_event(event)
```

After the `send_event(event)` I get the following error:
```
Error while sending event to Stream Machine (https://in.strm.services/event),
    response status = 400, response: Invalid event contract: . Not supported.
```
We need to set `strmMeta.eventContractRef` to a valid contract reference (`strm list event-contracts`). While a _schema_ defines the shape/structure of an event, an _event-contract_ defines the rules that apply to an event, i.e. what validations, what fields contain personally identifiable data, which field ties the events into a sequence etc...  We'll use `streammachine/example/1.3.0` (inspect with `strm get event-contract ..`)


```python
event.strmMeta.eventContractRef="streammachine/example/1.3.0"
sender.send_event(event)
Error while sending event to Stream Machine (https://in.strm.services/event),
    response status = 400, response: Field: 'consistentValue' in event with schema: 'streammachine/demo/1.0.2'
    with value: '' doesn't match regex: '^.+$'
```

So the field `consistentValue` needs at least one character?

```python
event.consistentValue="hi there"
sender.send_event(event)
# nothing ....
```
A whole lot of nothing is returned. Stream Machine is meant for very high throughput and logging anything at thousands of events per second will quickly break the bank if you're using StackDriver. So None and http status 204 are the indicators that everything went fine.

But if we have the `strm egress demo` running, we would see something like

```
strm egress demo
{
  "strmMeta": {
    "eventContractRef": "streammachine/example/1.3.0",
    "nonce": -1515353731,
    "timestamp": 1628688372401,
    "keyLink": "cd181172-4ec7-4f0d-86bb-662fc0ee854b",
    "billingId": "strmbart5986941267",
    "consentLevels": []
  },
  "uniqueIdentifier": null,
  "consistentValue": "AWpvnLU8hBPWRfYrAjWPs0wWt6vBMMXXEnSqGTw=",
  "someSensitiveValue": null,
  "notSensitiveValue": null
}

```
We can observe that the consistentValue field has some data in it (it's actually base64 encoded encrypted "hi there"). Explanation of the various fields in `strmMeta` are explained in the [documentation](https://docs.streammachine.io)

Let's fill in some more fields:
```python
import random, uuid

def create_avro_event():
    event = DemoEvent()

    event.strmMeta.eventContractRef = "streammachine/example/1.3.0"
    event.strmMeta.consentLevels = [random.randint(0, 3)]

    event.uniqueIdentifier = str(uuid.uuid4())
    event.someSensitiveValue = "A value that should be encrypted"
    event.consistentValue = "a-user-session"
    event.notSensitiveValue = "Anyone is free to see this text."
    return event

event = create_avro_event()

r = sender.send_event(event)
print(r)
None
```
In the `strm egress demo` pane we see:
```
{
  "strmMeta": {
    "eventContractRef": "streammachine/example/1.3.0",
    "nonce": 1820364498,
    "timestamp": 1628689078226,
    "keyLink": "5074c4de-c51b-4321-a70b-c87db4c79bde",
    "billingId": "strmbart5986941267",
    "consentLevels": [ 0 ]
  },
  "uniqueIdentifier": "AUvu95+NUDFf9krvvVUSU+pJsRBl9XahrMVCTpjqHDa9lTHBTzbRdjazyyMVi3xDy2Ps7HDxJHWA",
  "consistentValue": "AUvu958J4Lf8JWlxwEfdMXXSZpjxdkBSL4hl8Tk5MVHp3L4=",
  "someSensitiveValue": "AUvu95+bq6bw4Z1l9pTYLNQwd/ecdtntrH5mcBJNWv8n6n9jzYxKwEuSDUjig5lPNYqpZpU=",
  "notSensitiveValue": "Anyone is free to see this text."
}
```
Just looking at it, I can see that all the PII field in the event-contract have been encrypted

## Decrypting
We can create a _decrypted stream_ that *does* contain personal data.
```
strm create stream --derived-from demo --levels 2 --save
```
Start sending data in a loop
```
import time
while True:
    time.sleep(0.1)
    sender.send_event(create_avro_event())
```
And observe in the egress:
```
strm egress demo-2
{
  "strmMeta": {
    "eventContractRef": "streammachine/example/1.3.0",
    "nonce": 1165221388,
    "timestamp": 1628689364936,
    "keyLink": "5074c4de-c51b-4321-a70b-c87db4c79bde",
    "billingId": "strmbart5986941267",
    "consentLevels": [
      3
    ]
  },
  "uniqueIdentifier": "279e9dcb-a9a5-497b-8d46-b213106f7fab",
  "consistentValue": "a-user-session",
  "someSensitiveValue": "AUvu95+bq6bw4Z1l9pTYLNQwd/ecdtntrH5mcBJNWv8n6n9jzYxKwEuSDUjig5lPNYqpZpU=",
  "notSensitiveValue": "Anyone is free to see this text."
}
```
Note that `someSensitiveValue` is still encrypted (observe the base64 encoding).
This is because
* we're asking for a stream with consent-levels up to 2.
* the data owner has given consent level 3, but we don't need it, didn't ask for
  it and so we don't get it. If we'd asked for level 3 this would also have been
  decrypted.

If we observe the consent levels in the decrypted stream:
```
strm egress demo-2 | jq -c .strmMeta.consentLevels
[3]
[3]
[2]
[2]
```
we only see `[3]` and `[2]`. Events with consent levels lower than these
don't get exported to this decrypted stream.

## Conclusions

We can send data into Stream Machine, and receive them privacy safe and of guaranteed quality.

The next post in this series will show how to get all these data into a Google Cloud bucket, and how to use those data in a Jupyter notebook.