Mapping¶

Understanding the mapping process¶

Before you dive in to creating your own mappings, it is important to understand a little bit about how a mapping is actually performed. The most notable thing to keep in mind is that a mapping script that you provide is not evaluated at request time for each event. Instead a mapping is evaluated only once during startup and declares how the actual mapping should take place.

Schema evolution and default values¶

Schema evolution is the process of changing the schema over time as requirements change. For example when a new feature is added to your website, you add additional fields to the schema that contain specific information about user interactions with this new feature. In this scenario, you would update the schema to have these additional fields, update the mapping and then run Divolte Collector with the new schema and mapping. This means that there will be a difference between data that was written prior to the update and data that is written after the update. Also, it means that after the update, there can still be consumers of the data that still use the old schema. In order to make sure that this isn’t a problem, the readers with the old schema need to be able to read data written with the new schema and readers with the new schema should also still work on data written with the old schema.

Luckily, Avro supports both of these cases. When reading newer data with an older schema, the fields that are not present in the old schema are simply ignored by the reader. The other way around is slightly trickier. When reading older data with a new schema, Avro will fill in the default values for fields that are present in the schema but not in the data. This is provided that there is a default value. Basically, this means that it is recommended to always provide a default value for all your fields in the schema. In case of nullable fields, the default value could just be null.

One other reason to always provide a default value is that Avro does not allow to create records with missing values if there are no default values. As a result of this, fields that have no default value always must be populated in the mapping, otherwise an error will occur. This is problematic if the mapping for some reason fails to set a field (e.g. because of a user typing in a non-conforming location in the browser).

In addition to introducing new fields with defaults, other forms of changes such as renaming and type changes can be permitted under some circumstances. For full details on the changes that are permitted and how the writing and reading schemas are reconciled refer to the Avro documentation on schema resolution.

Schema evolution with Kafka and Google Cloud Pub/Sub¶

Schema evolution is not automatically available when using event-based delivery via Kafka or Google Cloud Pub/Sub. The file implementation is defined by the Avro project, but this is not suitable for messaging systems and no alternative is defined. As a result there is no single standard and various strategies are in use.

For Kafka there are two methods:

• Simply Avro-encode the record as a message payload. This is the naked (and default) mode in which Kafka sinks operate. Consumers are expected to implicitly know the schema used to write the record. This makes changing the schema infeasible without also switching to a new Kafka topic; consumers generally have no way to know which messages were written with the old or new schema.
• Tag the message in some way so that consumers can detect the schema used to encode the payload. The Confluent platform uses this approach, prefixing each message with a header to inform consumers which schema was used to write the message. Schema changes are possible so long as they remain backwardly compatible. Kafka sinks have experimental support for this, available by configuring the sink to operate in confluent mode.

For Google Cloud Pub/Sub the message data is the Avro-encoded record, identical to the naked mode for Kafka sinks. In addition to the data messages have a schemaFingerprint attribute that refers to the schema that was used to encode the message. (See Google Cloud Pub/Sub Sinks for details.) Schema changes are possible so long as they remain backwardly compatible.

Built-in default mapping¶

Divolte Collector comes with a built-in default schema and mapping. A mapping will use these if the mapping schema or script file are not specified. The default mapping is intended to map events from a browser source and will map most things that you would expect from a clickstream data collector. The Avro schema that is used can be found in the divolte-schema Github repository. The following mappings are present in the default mapping:

The default schema is not available as a mapping script. Instead, it is hard coded into Divolte Collector. This allows Divolte Collector to do something useful out-of-the-box without any complex configuration.

Values, fields and mappings¶

Mapping involves three main concepts: values, fields and mappings.

A value is something that is extracted from the incoming event (e.g. the location or a HTTP header value) or is derived from another value (e.g. a query parameter from the location URI). Values in the mapping are produced using calls to functions that are built into the mapping DSL. Below is the complete documentation for all values that can be produced. One example of such a function call would be calling location() for the location value or referer() for the referrer value of the event.

A field is a field in the Avro record that will be produced as a result of the mapping process. The type of a field is defined by the Avro schema that is used. Mapping is the process of mapping values extracted from the event onto fields in the Avro record.

A mapping is the piece that tells Divolte Collector which values need to be mapped onto which fields. The mapping DSL has a built in construct for this, explained below.

map location() onto 'locationField'

map {
  def u = parse location() to uri          // Parse the URI out of the location
  parse u.query().value('n') to int32      // Take the n query parameter and try to parse an int out of it
} onto 'intField'

map 'string literal' onto 'stringField'
map true onto 'booleanField'

when referer().isAbsent() apply {           // Only apply this mapping when a referer is absent
  map true onto 'directTraffic'
}

def u = parse referer() to uri              // parse a URI out of the referer
def q = u.query()                           // parse the query string of the URI
def i = parse q.value('foo') to int32       // parse a int out of the query parameter 'foo'
map i onto 'intField'                       // map it onto the field 'intField'

{ "name": "fieldName", "type": ["null","string"], "default": null }

{ "name": "fieldName", "type": ["null","boolean"], "default": null }

{ "name": "fieldName", "type": ["null","int"], "default": null }

{ "name": "fieldName", "type": ["null","long"], "default": null }

{ "name": "fieldName", "type": ["null","float"], "default": null }

{ "name": "fieldName", "type": ["null","double"], "default": null }

{
  "name": "fieldName",
  "type": [
    "null",
    {
      "type": "map",
      "values": {
        "type": "array",
        "items": "string"
      }
    }
  ],
  "default": null
}

{
  "name": "fieldName",
  "type":
    [
      "null",
      {
        "type": "array",
        "items": "int"
      }
    ],
  "default": null
}

{ "name": "fieldName", "type": ["null","bytes"], "default": null }

def i = parse stringValue to int32

def u = parse referer() to uri              // u is of type URI (which is not mappable)
def q = u.query()                           // q is of type map<string,list<string>>
def s = q.value('foo')                      // s is of type string if query parameter foo contained a integer number
def i = parse s to int32                    // i is of type int
map i onto 'intField'                       // map it onto the field 'intField'

{
  "total_price": 184.91,
  "items": [
    { "sku": "0886974140818", "count": 1, "price_per": 43.94 },
    { "sku": "0094638246817", "count": 1, "price_per": 22.99 },
    { "sku": "0093624979357", "count": 1, "price_per": 27.99 },
    { "sku": "8712837825207", "count": 1, "price_per": 89.99 }
  ]
}

{
  "type": [
    "null",
    {
      "name": "ShoppingBasket",
      "type": "record",
      "fields": [
        { "name": "total_price", "type": "float" },
        {
          "name": "items",
          "type": {
            "type": "array",
            "items": {
              "type": "record",
              "name": "LineItem",
              "fields": [
                { "name": "sku",       "type": "string" },
                { "name": "count",     "type": "int"    },
                { "name": "price_per", "type": "double" }
              ]
            }
          }
        }
      ]
    }
  ],
  "default": null
}

map digest('sha-256', 'pseudoSession')
    .add(partyId())
    .add('-')
    .add(sessionId())
    .result().toBase64() to 'pseudoSession'

when conditionBooleanValue apply {
  // Conditional mapping go here
  map 'value' onto 'fieldName'
}

when referer().isAbsent() apply {
  map true onto 'directTraffic'
}

when {
  def u = parse location() to uri
  u.query().value('clientId').isPresent()
} apply {
  map true onto 'signedInUser'
}

section {
  // Section's mappings go here
  map 'value' onto 'field'
}

section {
  def u = parse location() to uri

  when u.path().equalTo('/home.html') apply {
    map 'homepage' onto 'pageType'
    exit()
  }

  when u.path().equalTo('/contact.html') apply {
    map 'contactpage' onto 'pageType'
    exit()
  }

  map 'other' onto 'pageType'
}

// other mappings here

when referer().isAbsent() exit()

when referer().isAbsent() {
  stop()
}

when referer().isAbsent stop()

A word on groovy¶

Groovy is a dynamic language for the JVM. This means, amongst other things, that you don’t have to specify the types of variables:

def i = 40
println i + 2

The above snippet will print out 42 as you would expect. Note two things: we never specified that variable i is an int and also, we are not using any parentheses in the println function call. Groovy allows to leave out the parentheses in most function and method calls. The code above is equivalent to this snippet:

def i = 42
println(i + 2)

This in turn is equivalent to this:

def i = 42
println(i.plus(2))

This works well when chaining single argument methods. However, this can be more problematic with nested method calls. Suppose we have a function called increment(x) which increments the x argument by 1, so increment(10) will return 11. The following will not compile:

println increment 10

However this will:

println(increment(10))

Yet this won’t:

println(increment 10)

In the Divolte Collector mapping DSL, it is sometimes required to chain method calls. For example when using the result of a casting operation in a mapping. We solve this by accepting a closure that produces a value as result:

map { parse cookie('customer_id') to int32 } onto 'customerId'

This way you don’t have to add parentheses to all intermediate method calls and we keep the syntax fluent. If you follow these general guidelines, you should be safe:

• When calling methods that produce a value, always use parentheses. For example: location(), referer(), partyId()

• When deriving a condition or other value from a method that produces a value, also use parentheses. For example:

  when location().equalTo('http://www.example.com/') apply {
    ...
  }
  
  map cookie('example').isPresent() onto 'field'
  
  map parsedUri.query().value('foo') onto 'field'
  

• When parsing or matching on something, extract it to a variable before using it. This also improves readability:

  def myUri = parse location() to uri
  when myUri.query().value('foo').isPresent() apply { ... }
  
  def myMatcher = match '^/foo/bar/([a-z]+)/' against myUri.path()
  when myMatcher.matches() apply { ... }
  

• When casting inline, use the closure syntax for mapping or conditionals:

  map { parse cookie('example') to int32 } onto 'field'
  

map location() onto 'locationField'

map referer() onto 'refererField'

map firstInSession() onto 'first'

map corrupt() onto 'detectedCorruption'

map duplicate() onto 'detectedDuplicate'

map timestamp() onto 'timeField'

map clientTimestamp() onto 'timeField'

map remoteHost() onto 'ipAddressField'

map viewportPixelWidth() onto 'widthField'

map viewportPixelHeight() onto 'widthField'

map screenPixelWidth() onto 'widthField'

map screenPixelHeight() onto 'widthField'

map devicePixelRatio() onto 'ratioField'

map partyId() onto 'partyField'

map sessionId() onto 'sessionField'

map pageViewId() onto 'pageviewField'

map eventId() onto 'eventField'

map userAgentString() onto 'uaField'

map cookie('cookie_name') onto 'customCookieField'

map eventType() onto 'eventTypeField'

// In the JavaScript
divolte.signal('myEvent', { foo: 'hello', bar: 42 });

{
  "eventType": "myEvent",
  "parameters": {
    "foo": "hello",
    "bar": 42
  }
}

map eventParameters() onto 'parametersField'

// In the JavaScript
divolte.signal('myEvent', { foo: 'hello', bar: 42 });

{
  "eventType": "myEvent",
  "parameters": {
    "foo": "hello",
    "bar": 42
  }
}

map eventParameters().value('foo') onto 'fooField'

// Or with a cast:
map { parse eventParameters().value('bar') to int32 } onto 'barField'

// In the JavaScript
divolte.signal('searchResults', [
  { "sku": "0886974140818", "score": 0.9 },
  { "sku": "0094638246817", "score": 0.8 }
]);

{
  "eventType": "searchResults",
  "parameters": [
    { "sku": "0886974140818", "score": 0.9 },
    { "sku": "0094638246817", "score": 0.8 }
  ]
}

{
  "name": "searchResults",
  "type": [ "null", { "type": "array", "items": "string" } ],
  "default": null
}

map eventParameters().path('$[*].sku') onto 'searchResults'

def locationUri = parse location() to uri

def locationUri = parse location() to uri
map locationUri.path() onto 'locationPathField'

def locationUri = parse location() to uri
map locationUri.rawPath() onto 'locationPathField'

def locationUri = parse location() to uri
map locationUri.scheme() onto 'locationSchemeField'

// or check for HTTPS and map onto a boolean field
map locationUri.scheme().equalTo('https') onto 'isSecure'

def locationUri = parse location() to uri
map locationUri.host() onto 'locationHostField'

def locationUri = parse location() to uri
map locationUri.port() onto 'locationPortField'

def locationUri = parse location() to uri
map locationUri.decodedQueryString() onto 'locationQS'

def locationUri = parse location() to uri
map locationUri.rawQueryString() onto 'locationQS'

def locationUri = parse location() to uri
map locationUri.decodedFragment() onto 'locationFragment'

def locationUri = parse location() to uri
map locationUri.rawFragment() onto 'locationFragment'

// If location() = 'http://www.example.com/foo/#/local/path/?q=hello+world'
// this would map '/local/path/' onto the field clientSidePath
def locationUri = parse location() to uri
def localUri = parse location().rawFragment() to uri
map localUri.path() onto 'clientSidePath'

def locationUri = parse location() to uri
def locationQuery = locationUri.query()
map locationQuery onto 'locationQueryParameters'

{
  "name": "uriQuery",
  "type": [
    "null",
    {
      "type": "map",
      "values": {
        "type": "array",
        "items": "string"
      }
    }
  ],
  "default": null
}

def locationUri = parse location() to uri
def locationQuery = locationUri.query()
map locationQuery.value('foo') onto 'fooQueryParameter'

def locationUri = parse location() to uri
def locationQuery = locationUri.query()
map locationQuery.valueList('foo') onto 'fooQueryParameterValues'

def matcher = match '/foo/bar/([a-z]+).html$' against location()

def locationUri = parse location() to uri
def pathMatcher = match '^/foo/bar/([a-z]+).html$' against locationUri.path()
when pathMatcher.matches() apply {
  map 'fooBarPage' onto 'pageTypeField'
  map pathMatcher.group(1) onto 'pageNameField'
}

def matcher = match '^/foo/bar/([a-z]+).html$' against location()

// use in conditional mapping
when matcher.matches() apply {
  map 'fooBarPage' onto 'pageTypeField'
}

// or map directly onto a boolean field
map matcher.matches() onto 'isFooBarPage'

// Using group number
def matcher = match '/foo/bar/([a-z]+).html$' against location()
map matcher.group(1) onto 'pageName'

// Using named capture groups
def matcher = match '/foo/bar/(?<pageName>[a-z]+).html$' against location()
map matcher.group('pageName') onto 'pageName'

map header('header-name') onto 'fieldName'

{
  "name": "headers",
  "type":
    [
      "null",
      {
        "type": "array",
        "items": ["string"]
      }
    ],
  "default": null
}

map header('header-name').first() onto 'fieldName'

map header('header-name').last() onto 'fieldName'

map header('header-name').get(2) onto 'fieldName'

map header('header-name').commaSeparated() onto 'fieldName'

// A simple digester.
def digester = digest("sha-256")

// A seeded digester.
def digester = digest("sha-256","a seed value")

map digest('sha-256', 'pseudoSession')
    .add(partyId())
    .add('-')
    .add(sessionId())
    .result() onto 'fieldName'

map digest('sha-256', 'aSeed')
    .add('...')
    .result() onto 'fieldName'

def ua = userAgent()

map userAgent().name() onto 'uaNameField'

map userAgent().family() onto 'uaFamilyField'

map userAgent().vendor() onto 'uaVendorField'

map userAgent().type() onto 'uaTypeField'

map userAgent().version() onto 'uaVersionField'

map userAgent().deviceCategory() onto 'uaDeviceCategoryField'

map userAgent().osFamily() onto 'uaOSFamilyField'

map userAgent().osVersion() onto 'uaOSVersionField'

map userAgent().osVendor() onto 'uaOSVendorField'

// uses the remoteHost as IP address to lookup
def ua = ip2geo()

// If a load balancer sets custom headers for IP addresses, use like this
def ip = header('X-Forwarded-For').first()
def myUa = ip2geo(ip)

// Returns the last item of the list
def ip = header('X-Forwarded-For').last()
// Returns the second last item
def ip = header('X-Forwarded-For').get(-2)

map ip2geo().cityId() onto 'cityIdField'

map ip2geo().cityName() onto 'cityNameField'

map ip2geo().continentCode() onto 'continentCodeField'

map ip2geo().continentId() onto 'continentIdField'

map ip2geo().continentName() onto 'continentNameField'

map ip2geo().countryCode() onto 'countryCodeField'

map ip2geo().countryId() onto 'countryIdField'

map ip2geo().countryName() onto 'countryNameField'

map ip2geo().latitude() onto 'latitudeField'

map ip2geo().longitude() onto 'longitudeField'

map ip2geo().metroCode() onto 'metroCodeField'

map ip2geo().timeZone() onto 'timeZoneField'

map ip2geo().mostSpecificSubdivisionCode() onto 'mostSpecificSubdivisionCodeField'

map ip2geo().mostSpecificSubdivisionId() onto 'mostSpecificSubdivisionIdField'

map ip2geo().mostSpecificSubdivisionName() onto 'mostSpecificSubdivisionNameField'

map ip2geo().postalCode() onto 'postalCodeField'

map ip2geo().subdivisionCodes() onto 'subdivisionCodesField'

map ip2geo().subdivisionIds() onto 'subdivisionIdsFields'

map ip2geo().subdivisionNames() onto 'subdivisionNames'