Loader ME to CDP

Loader ME to CDP extracts raw customer events from the Meiro Events (ME) database, transforms the payload based on payload transformation definitions, and uploads the transformed customer events to a CDP database. 

Data In/Data Out

Learn more: about the folder structure here.

Parameters

Credentials

Meiro Events Database

Meiro CDP Login

Example of valid configurations:

{
    "parameters": {
        "credentials": {
            "me_db": {
                "host": "me.meiro.io",
                "port": "1234",
                "user": "meiro",
                "database": "meiro_events",
                "#password": "sessame"
            },
            "cdp": {
                "url": "https://cdp.meiro.io/",
                "username": "cdp@meiro.io",
                "#password": "sessame"
            }
        },
        "dry_run": false,
        "first_run_time_window": "30 seconds ago",
      "events": []
    }
    }

Events

Below is an array of definitions used for the extraction raw customer events from the ME database and transformation of customer event payload.

Example of the event definition without unnest_payload_array specification:

{
  "enabled": true,
  "schema": "events",
  "table_name": "events",
  "source_id": "me_web",
  "event_type": "me_page_view",
  "version": "1-0-0",
  "event_time": "payload ->> 'timestamp'",
  "user_identifier": "payload ->> 'user_id'",
  "event_filter": "payload ->> 'type' = 'page_view'",
  "payload": "payload",
  "payload_transformation_definitions": []
}

WITH events_events AS (
    SELECT
            reception_time as "reception_time",
            payload as "payload"
    FROM events.events
    WHERE reception_time >= {{some_timestamp}}::timestamp
)
SELECT
        'me_web' as "source_id",
        'me_page_view' as "type",
        '1-0-0' as "version",
        payload ->> 'timestamp' as "event_time",
        payload ->> 'user_id' as "user_identifier",
        payload as "payload"
FROM events_events
WHERE payload ->> 'type' = 'page_view'

Example of the event definition with unnest_payload_array specification:

{
  "enabled": true,
  "schema": "events",
  "table_name": "magnews_events",
  "source_id": "mn",
  "event_type": "mn_sent",
  "version": "1-0-0",
  "event_time": "reception_time",
  "user_identifier": "payload ->> 'MESSAGE.TO'",
  "event_filter": "payload ->> 'EVENT.TYPE' = 'Sent'",
  "payload": "payload",
  "unnest_payload_array": "payload -> 'json' -> 'data'",
  "payload_transformation_definitions": []
}

WITH events_magnews_events AS (
    SELECT
            reception_time as "reception_time",
            payload as "payload"
    FROM events.magnews_events
    WHERE reception_time >= {{some_timestamp}}::timestamp
),
unnested AS (
  SELECT 
    reception_time as "reception_time", 
    jsonb_set(
      jsonb_set(
        jsonb_set(
          jsonb_set(
            jsonb_set(
              jsonb_set(
                jsonb_array_elements(payload -> 'json' -> 'data'), 
                '{method}', 
                payload -> 'method'
              ), 
              '{cookies}', 
              payload -> 'cookies'
            ), 
            '{headers}', 
            payload -> 'headers'
          ), 
          '{endpoint}', 
          payload -> 'endpoint'
        ), 
        '{parsed_url}', 
        payload -> 'parsed_url'
      ), 
      '{remote_addr}', 
      payload -> 'remote_addr'
    ) as "payload" 
  FROM 
    events_magnews_events 
  WHERE 
    "reception_time" >= '2023-01-01 00:00:00' :: timestamp
) 
SELECT 
  'mn' as "source_id", 
  '1-0-0' as "version", 
  'mn_sent' as "type", 
  (reception_time):: timestamp as "event_time", 
  (payload ->> 'MESSAGE.TO'):: text as "user_identifier", 
  payload as "payload" 
FROM 
  unnested 
WHERE 
  payload ->> 'EVENT.TYPE' = 'Sent'

Transformation definitions

This section describes how the extracted raw customer events undergo transformation using JSON payloads. Key features of transformation:

1. The order of transformations is important, as each transformation is applied one by one to the payload in the respective order.
2. All JSON (payload) paths must be defined using dot notation (e.g., payload.payload.email).
3. The component supports data transformation only in key: value object. It does not support analyzing and iterating throw array objects.
4. The user must ensure that a new payload path is possible to create. Example of the newly created path:

It is not possible to create new_path_incorrect, because email key is already occupied by string object. It is not possible to create another nested object, and the component ends up with an error.

Allowed payload transformations

The ME to CDP component supports several transformations, which might be used in any order and as many times as you want to reach the required result of the payload. All transformations could be disabled using: 

{
  "enabled": false
}

Rename key

This feature allows the component to extract a value from a specified old path, create a new path, place the extracted value in the new path, and delete the old path. This helps organize and restructure the data within the payload, improving data quality and consistency.

Value mapping

The component first extracts the value from the target_old_value path. It then searches for a matching value in the value_maps array, comparing the extracted value with the old value from each value_map. If a match is found, the component takes the corresponding new value from the value_map, creates a new path named target_new_value, and places the new value from the value_map there. This process is similar to a join operation in a database.

Enrich pop up web banners

The component first extracts the banner_id value from  id_path. It then searches for matching banner_name from the refreshed web banner data downloaded from the CDP instance. If a match is found, the component creates a new path called name_path in the payload and places the banner_name value there.

Enrich embedded web banners

Same logic as in Enrich pop up web banners applied to payloads where embedded web banners data must be enriched.

Vocative from first name

This feature is designed for projects based in the Czech Republic, where vocative forms of names are commonly used. The component extracts the first name from the specified path using the first_name_path. Then tries to use vocative library to transform first_name. The transformed name is then placed in a new path specified by result_path.

Gender estimate from first name

This functionality is mainly for projects in the Czech Republic because it uses a vocative library that works mostly with first names in the Czech calendar. Component extracts first_name from first_name_path path. Then try to use vocative library to estimate gender from first_name. It creates a new path in result_path and places estimated gender there.

Example of "Gender estimate from first name"

Replace string

Component extracts string object from string_path path. Then apply python replace method on an extracted object with old and new as an argument. It creates a new path  result_path and places transformed string there.

Strip string

Component extracts string object from string_path path. Then apply the python replace method on an extracted object with old and new as an argument. It creates a new path  result_path and places transformed string there.

Lower string

Component extracts string object from string_path path. Then applies the python lower method on an extracted object without any arguments. It creates a new path in result_path and places transformed string there.

Validate email

Component extracts string object from string_path path. Then applies the python lower method on an extracted object without any arguments. It creates a new path in result_path and places transformed string there.

Delete from path

Component checks if delete_path exists and removes it.