Architecture

Center Replication

Realtime Replication

This document details how to replicate data from one AMI Center to another AMI Center using its built-in support for data replication. It supports subscription to tables and replication to another AMI Center.

This document provides an example on how to use AMI Center Replication.

Prerequisites

• Have at least 2 AMI centers running (e.g. host:center_port e.g. localhost:3270 & localhost:4270)

• Have the table schema predefined on both source and target centers.

Replication Procedures

__ADD_CENTER(CenterName String, Url String, Certfile String, Password String)

• CenterName (Required): Specified name for center (source)
• Url (Required): Url of center (host: ami.center.port)
• Certfile: Path to the certfile (optional)
• Password: Password if a certfile is given (optional)

__REMOVE_CENTER(CenterName String)

__ADD_REPLICATION(Definition String, Name String, Mapping String, Options String)

• Definition: Target_TableName=Source_CenterName.Source_TableName or Source_CenterName.TableName (if both source and target has same table name)
• Name: Name for the replication
• Mapping (Optional): Mappings to be applied for the tables, (key value delimited by comma) ex: "target_col_name=source_col_name" or ex: "(act=account,symbol=symbol,value=(px*qty))". Pass in null to skip mapping.
• Options (Optional): options to clear the replication on
  • "Clear=onConnect" or "Clear=onDisconnect" or "Clear=Off"
    • Note: If configured the replicated table will clear when the center connects or disconnect

__REMOVE_REPLICATION(Name String)

SHOW CENTERS

• Shows properties of all replication sources. Properties shown include:
  • CenterName (String)
  • URL (String)
  • CertFile (String)
  • Password (String)
  • Status (String): shows whether the replication source is connected or not
  • ReplicatedTables (int): Number of replicated tables
  • ReplicatedRows (int): Number of replicated rows

SHOW REPLICATIONS

• Shows properties of all replications. Properties shown include:
  • ReplicationName (String)
  • TargetTable (String)
  • SourceCenter (String)
  • SourceTable (String)
  • Mapping (String): shows the mapping for the replication
  • Options (String): shows options to clear replications on
  • ReplicatedRows (int): Number of replicated rows
  • Status (String): shows whether a successful connection is established to the target table

Note: When replicating from the source AMI Center and the source table, ensure the table is a Public table with Broadcast Enabled.

Note 2: Configuring the RefreshPeriodMs will allow you to adjust for throughput, performance and how often updates are pushed.

See the following documentation: CREATE PUBLIC TABLE

Lower RefreshPeriodMs means updates are pushed more frequently which potentially means lower throughput.

Higher RefreshPeriodMs could mean higher throughput and better performance but fewer updates.

Note 3: Removing the replication will clear all the copied entries on the target side's table, regardless of the options.

Replication Sample Guide

To replicate data from one AMI Center to a destination AMI Center, first we need to add the source AMI Center in the destination AMI Center. After which we can replicate target tables from the source to the destination.

call __ADD_CENTER("source", "localhost:3270");

__ADD_CENTER Adds the source center you want to copy from. So if your AMI Center Port is 3270, and you call __ADD_CENTER on the destination AMI Center, that means you plan to replicate data from 3270 to 4270.

call __ADD_REPLICATION("dest_table=source.mytable","myReplication"," account=account", "Clear=onConnect");

__ADD_REPLICATION specifies which table on the source side the target side wants to replicate. The data travels from source to target, from 3270 to 4270.

AMI Center Persist Files

There are two persist files responsible for maintaining the replications, __REPLICATION.dat and __CENTER.dat. We do not recommend modifying the contents of these files manually, these files may have strict formatting and could cause serious issues if altered incorrectly.

Persist File Location

By default these files are located in the default persist directory. This may not be the case if the persist dir has been changed by the property: ami.db.persist.dir or the system tables directory has been changed by the property: ami.db.persist.dir.system.tables.

To change only the __REPLICATION.dat and __CENTER.dat locations add the following property.

ami.db.persist.dir.system.table.__REPLICATION=path
ami.db.persist.dir.system.table.__CENTER=path

__REPLICATION.dat Format:

ReplicationName|TargetTable|SourceCenter|SourceTable|Mapping|Options| "ReplicationName"|"SourceTableName"|"SourceName"|"TargetTableName"|"TableMapping"|"ClearSetting"|

__CENTER.dat Format:

CenterName|Url|CertFile|Password| "CenterName"|"CenterURL"|"CertFileIfProvided"|"PasswordForCertFile"|

Daily Replication

In this example we will write a simple procedure to transfer historical data from one center to another.

In our working center we will use a table with the following schema;

CREATE PUBLIC TABLE DataTable (AccountID String, Price double, Quantity Double, Date Long, D long);

Similarly, in the center with historical data we have a table with the same name but a different schema:

CREATE PUBLIC TABLE DataTable (AccountID String, Price double, Quantity Double, Date Long, SourceD Long, HistDate Long);

NB: D is an auto-generated incrementing unique id for the row which unique across all tables (see Reserved Columns on Public Tables).

Next we will use the below procedure to transfer the historical data. The arguments for the procedure are as follows:

• tableName - this is the table you will be transferring data from and to (ie. DataTable)
• histCenter - this is the name of the datasource where the historical data will sent to
• whereClause - argument that can be used to get the data you want to transfer (ie. you may want to send across data where Date == 20220101)
• batchSize - argument to specify the number of data rows to send across in each go

Note that this procedure uses column D to decide which data to send.

CREATE PROCEDURE MoveRowsToHist OFTYPE AMISCRIPT USE 

  arguments="string tableName, string histCenter, string whereClause, int batchSize" 

  script="long histDate = formatDate(timestamp(), \"yyyyMMdd\", \"UTC\");
          long srcTblDmax = select max(D) from ${tableName} where ${whereClause};
          srcTblDmax = srcTblDmax != null ? srcTblDmax : 0L;
          int destTblDmax = use ds=${histCenter} execute select max(SourceD) from ${tableName} where ${whereClause} AND HistDate == ${histDate};
          destTblDmax = destTblDmax != null ? destTblDmax : 0;

          while (srcTblDmax > destTblDmax) {
              use ds=${histCenter} insert into ${tableName} from select * except(D), D as SourceD, ${histDate} as HistDate from ${tableName} 
                  where ${whereClause} AND D > destTblDmax limit batchSize order by SourceD;
              destTblDmax = use ds=${histCenter} execute select max(SourceD) from ${tableName} where ${whereClause} AND HistDate == ${histDate};
         }"

CREATE PROCEDURE MoveRowsToHist OFTYPE AMISCRIPT USE arguments="string tableName, string histCenter, string whereClause, int batchSize" script="long histDate = formatDate(timestamp(), \"yyyyMMdd\", \"UTC\"); long srcTblDmax = select max(D) from ${tableName} where ${whereClause}; srcTblDmax = srcTblDmax != null ? srcTblDmax : 0L; int destTblDmax = use ds=${histCenter} execute select max(SourceD) from ${tableName} where ${whereClause} AND HistDate == ${histDate}; destTblDmax = destTblDmax != null ? destTblDmax : 0; while (srcTblDmax > destTblDmax) {use ds=${histCenter} insert into ${tableName} from select * except(D), D as SourceD, ${histDate} as HistDate from ${tableName} where ${whereClause} AND D > destTblDmax limit batchSize order by SourceD; destTblDmax = use ds=${histCenter} execute select max(SourceD) from ${tableName} where ${whereClause} AND HistDate == ${histDate};}"