Ambar User Guide

Ambar supports MySQL servers running versions greater than or equal to 5.7.9 (for the 5.7 series), and greater than or equal to 8.0.11 (for the 8.0 series).

The following configuration items are required:

binlog_format = row
log_bin = /var/lib/mysql/mysql-bin.log
expire_logs_days = 7

On RDS, retention (expire_log_days) is set with the following command:

CALL mysql.rds_set_configuration('binlog retention hours', 24);

Creation of a user with the appropriate grants can be performed with the following sequence of commands:

CREATE USER 'projections_user'@'%' IDENTIFIED BY 'PASSWORD_GOES_HERE_CHANGE_ME';
GRANT REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'projections_user'@'%';
GRANT SELECT ON *.* TO 'projections_user'@'%';
GRANT SELECT ON mysql.rds_heartbeat2 TO 'projections_user'@'%';
GRANT SELECT ON mysql.rds_configuration TO 'projections_user'@'%';
GRANT EXECUTE ON PROCEDURE mysql.rds_kill TO 'projections_user'@'%';

For information about the binary log and row based logging, please see the following links: row-based-logging, binary-log.

For information about binary log retention settings, please see the following link: binary-log-retention.

For information about binlog synchronization and InnoDB flushing, please see the following links: binlog-synchronization, InnoDB-flushing.

For information about replication privileges, please see the following link: replication-privileges.

PostgreSQL servers greater than or equal to version 14 are supported.

For information about configuration for logical replication, see logical-replication-config.

For more information about creating publications, see publication-creation.

For more information about required privileges, see required-privileges.

MySQL Type	Ambar Type
Bool	Boolean
Serial	Unsigned Integer
TinyInt, SmallInt, MediumInt, BigInt, Int	Signed Integer
Float, Double	Real
UTF8 Char types	String
Bit, Binary, Varbinary	Byte String
Blob, TinyBlob, MediumBlob, LongBlob	Byte String
Json	JSON
DateTime	DateTime

Note: UTF8 Char types include: Char, NChar, Varchar, NVarchar, Text, TinyText, MediumText, and LongText columns with an encoding of UTF8MB4, UTF8, or UTF8MB3.

PostgreSQL Type	Ambar Type
int2, int4, int8	Signed Integer
float4, float8	Real
bool	Boolean
json	JSON
bytea	Bytes
timestamp, timestamptz	DateTime
text	String

The filter DSL supports the following boolean literals: true and false.

The filter DSL supports numeric literals in both integer and scientific format. Examples include 3, -4, 5.0, and 3.14e5.

The filter DSL supports string literals, delimited by double quotation marks. For example: "hello world".

Finally, it supports the null value literal, denoted null.

The filter DSL supports a standard set of logical connective operators: not for logical negation, && for logical conjunction, and || for logical disjunction.

It also supports generic equality operators, denoted = and != that work for strings and numbers.

A set of comparison operators is available for numeric values: <, >, <=, >=.

Two string operators are available: substring(<haystack>, <needle>), and regex_match(<string>, <regex>). The first of these checks whether the <needle> argument is a substring of the <haystack> argument, and the second checks if the <string> matches the <regex>. The filter DSL supports extended Posix regular expressions.

Two lookup operators are available to extract values from the record under test. The first, denoted lookup(<field_name>) takes a string literal as an argument, and evaluates to the value of the indicated field in the record under test.

There is also a JSON lookup operator, denoted lookup_json(<field_name>, <json_path>). This takes two string literals as arguments, one indicating a field, and one indicating a JSON path. It evaluates to the value stored at that path in the JSON string stored under the indicated field name in the record.

Parentheses can be used to alter the default precedence rules, which are standard unless otherwise noted.

Paths into JSON objects are specified as a sequence of dot separated path elements. Each path element is either a field lookup in an object, or an index into an array. Object field lookups are specified using the name of the field, and array indices are specified with natural numbers inside square brackets.

The reserved characters in these paths are: ., [, ], \, and '. When these characters are required outside of their special meaning, they can be escaped with a double backslash: \\. The path element separator is optional prior to array index path elements. Single quotes can be optionally used to delimit object indices.

In the examples that follow, suppose we have a data source which contains records having five fields:

• seqid storing an unsigned integer
• auto_increment storing an unsigned integer
• timestamp storing a datetime value
• value storing a json object
• full_name storing a customer full name

In the JSON encoding for delivery, these records would have a payload value that looks like this:

{
  "seqid": 1,
  "auto_increment": 2,
  "timestamp": "2024-02-19T18:29:12Z",
  "value": "{"x": 1, "y": 2}",
  "full_name": "John Doe"
}

The simplest filter is the one which selects all records. That filter is specified using the boolean literal value, as below:

true

A filter which selects only records from the sequence with id 3 would look like:

lookup("seqid") == 3

A filter which selects only records produced on the United States Independence Day, 2024 would look like:

substring(lookup("timestamp"), "2024-07-04")

A filter which selects records whose value column has an x field less than 10 would look like:

lookup_json("value", ".x") <= 10

A filter which matches records where the full_name field contains the last name "Doe" would look like:

regex_match(lookup("full_name"), " Doe$")

Ambar requires responses from the endpoint to take on one of three forms depending on whether the associated record was correctly processed. In the event of successful record processing, the endpoint should respond with the following JSON body:

{
  "result": {
    "success": {}
  }
}

In the event of an inability to successfully process the message, you can request Ambar either ignore the failure and continue delivering the remainder of the record sequence, or you can request that Ambar pause delivering the remainder of the record sequence, and attempt redelivery at a later time.

To request that the failure be ignored and the remainder of the record sequence be processed, your endpoint should respond with the following JSON body:

{
  "result": {
    "error": {
      "policy": "keep_going", 
      "class": "keep it going", 
      "description": "keep it going"
    }
  }
}

To request that delivery be paused and delivery of the failed record re-attempted, your endpoint should respond with the following JSON body:

{
  "result": {
    "error": {
      "policy": "must_retry", 
      "class": "must retry it", 
      "description": "must retry it"
    }
  }
}

In either case, the class and description fields can be filled in however you choose.

Ambar guarantees that each record sequence in your data sources will be filtered and transmitted in order, at least once to each of your data destinations. If you require exactly once processing, your HTTP endpoint will need to deduplicate messages. You must deduplicate through a field or combination of fields that is universally unique. For example a message UUID, or a combination of a data source id and an auto incrementing column.

Upon deduplicating a record in this way, please issue a success response as described above.

In the following sections, we list each of the endpoints of the API, along with their associated request formats. For detailed information about their response formats, please refer to our OpenAPI spec.

{
  "dataSourceType": "postgres",
  "dataSourceConfig": {
    "hostname": "db.host.com",
    "hostPort": "5432",
    "databaseName": "my-db",
    "tableName": "my-table",
    "publicationName": "my-publication",
    "columns": "serial_col,partitioning_col,value",
    "username": "replication-user",
    "password": "replication-user-pw",
    "serialColumn": "serial_col",
    "partitioningColumn": "partitioning_col",
    "tlsTerminationOverrideHost": "my.override.host"
  }
}

{
  "dataSourceType": "mysql",
  "dataSourceConfig": {
    "hostname": "db.host.com",
    "hostPort": "3306",
    "databaseName": "my_db",
    "tableName": "my_table",
    "columns": "incrementing_col,partitioning_col,value",
    "binlogReplicationServerId": "25039"
    "username": "replication-user",
    "password": "replication-user-pw",
    "incrementingColumn": "incrementing_col",
    "partitioningColumn": "partitioning_col",
    "tlsTerminationOverrideHost": "my.override.host"
  }
}

{
  "resourceId": "AMBAR-dd5e24ba-f1da-4c96-a02c-e1c2e0407d05"
}

{
  "dataSourceId", "data-source-1",
  "filterContents", "..."
}

{
  "resourceId": "AMBAR-ffdbe21f-4888-4059-9503-d312305f2ee8"
}

{
    "filterIds": ["Filter-1", "Filter-2"]
    "destinationEndpoint": "stream.processing.example.com/abc",
    "username":"guinea",
    "password":"pig"
}

{
  "resourceId": "AMBAR-9d29e6fe-7316-479c-9e7b-1b49459efe76"
}

For the Ambar OpenAPI spec, please see the following link: ambar-openapi-spec.

For the Ambar Terraform provider, see the following link: ambar-terraform-provider.

For a detailed description of how to configure a data source, see the following link: data-source-config.

For a detailed description of how to configure a data destination, see the following link: data-destination-config.

For a detailed description of how to configure a filter, see the following link: filter-config.