NAME
incline - a replicator for RDB shards
SYNOPSIS
incline [options] command
DESCRIPTION
Incline is a replicator for MySQL / PostgreSQL with following
characteristics.
* replicates information within a single database node or between
database shards
* replication rules defined in JSON files
* synchronous replication within a single database through the use of
automatically-generated triggers
* asynchronous (eventually consistent) replication between database
nodes using automatically-generated queue tables and fault-torelant
forwarders
This manual consists of three parts, "INSTALLATION", "TUTORIAL",
"COMMAND REFENENCE", and "FILE FORMATS". For design documentation and
background knowledge, please refer to the URLs listed in the "SEE ALSO"
section.
INSTALLATION
Incline uses "autotools" and automatically tries to detect the client
libraries of MySQL and / or PostgreSQL, so a typical installation
procedure will be as follows.
% ./configure
% make
# make install
If configure fails to locate the client libraries, --with-mysql and
--with-pgsql options can be used.
% ./configure --with-mysql=my_mysql_installation_dir
Also, if you have perl and its DBI drivers installed, it is possible to
run the embedded tests using make.
% make test
TUTORIAL
The tutorial explains how to create a microblog service (like twitter)
running on four database shards. Incline (by itself) does not support
adding database nodes without stopping the service. If you are
interested in such feature, please refer to the documentation of
"Pacific" after reading this tutorial.
CREATING TABLES
At least four tables are needed to create a microblog service on
database shards. Instead of a single table representing follower <=>
followee relationship, each user needs to have a list of followers (or
list of following users) to him / her on his / her database shard. Also,
each user need to have his / her `timeline' table on his / her shard (or
else the service would not scale out). The example below is a minimal
schema on MySQL. All shards should have the same schema applied. Two
tables, `following' and `tweet' will be modified by the application.
`Follower' and `timeline' tables will be automatically kept (eventually)
in sync by incline with the former two tables.
CREATE TABLE following (
userer_id INT UNSIGNED NOT NULL,
following_id INT UNSIGNED NOT NULL,
PRIMARY KEY (user_id,following_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
CREATE TABLE follower (
user_id INT UNSIGNED NOT NULL,
follower_id INT UNSIGNED NOT NULL,
PRIMARY KEY (user,following_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
CREATE TABLE tweet (
tweet_id INT UNSIGNED NOT NULL AUTO_INCREMENT,
user_id INT UNSIGNED NOT NULL,
creation_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
body VARCHAR(255) NOT NULL,
PRIMARY KEY (tweet_id),
KEY (user_id,tweet_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
CREATE TABLE timeline (
user_id INT UNSIGNED NOT NULL,
tweet_user_id INT UNSIGNED NOT NULL,
tweet_id INT UNSIGEND NOT NULL,
creation_time TIMESTAMP NOT NULL,
PRIMARY KEY (user_id,creation_time,tweet_user_id,tweet_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
WRITING THE REPLICATION DEFINITION FILE
To keep `follower' and `timeline' tables in sync with the other two,
replication rules should be defined. The example below show the
definition corresponding to the table schema above.
The first hash defines how the `follower' tables should be kept
synchronized to the `following' tables. `Following_id' and `user_id'
columns of `following' tables are mapped to `user_id' and `follower_id'
columns of `follower' tables, and `follower' tables are sharded using
the `user_id' column.
The second hash defines how the `timeline' tables should be constructed
from the `follower' tables and `tweet' tables. In addition to the
definitions of `pk_columns' and `shard-key', `merge' property of the
hash defines how the two source tables should be merged (using INNER
JOIN).
[
{
"destination" : "follower",
"source" : "following",
"pk_columns" : {
"following.following_id" : "user_id",
"following.user_id" : "follower_id"
},
"shard-key" : "user_id"
},
{
"destination" : "timeline",
"source" : [ "follower", "tweet" ],
"pk_columns" : {
"follower.follower_id" : "user_id",
"tweet.user_id" : "tweet_user_id",
"tweet.tweet_id" : "tweet_id",
"tweet.creation_time" : "creation_time"
},
"merge" : {
"follower.user_id" : "tweet.user_id"
},
"shard-key" : "user_id"
}
]
WRITING THE SHARD DEFINITION FILE
Another definition file is required when using incline for synchronizing
database shards. The following example represents a distributed database
with four shards using range partitioning. First node with the IP
address 10.1.1.1 handles ids from 0 to 9999, second node (10.1.1.2)
handles 10000 to 19999, third (10.1.1.1.3) handles 20000 to 29999,
fourth (10.1.1.4) handles ids equal to or greater than 3000.
{
"algorithm" : "range-int",
"map" : {
"0" : [ {
"host" : "10.1.1.1"
} ],
"10000" : [ {
"host" : "10.1.1.2"
} ],
"20000" : [ {
"host" : "10.1.1.3"
} ],
"30000" : [ {
"host" : "10.1.1.4"
} ]
}
}
In addition to `range-int', `hash-int' algorithm is also supported. A
hash-based shard definition will look like below, you may use either one
to run the microblog service described in this tutorial.
{
"algorithm" : "hash-int",
"num" : 4,
"nodes" : [
[ {
"host" : "10.1.1.1"
} ],
[ {
"host" : "10.1.1.2"
} ],
[ {
"host" : "10.1.1.3"
} ],
[ {
"host" : "10.1.1.4"
} ]
]
}
INSTALLING QUEUE TABLES AND TRIGGERS
The next step is to install triggers and to create queue tables using
the definitions files. The following commands create queue tables and
installs triggers on the database running on 10.1.1.1. The commands
should be applied to all of the database shards.
% incline --rdbms=mysql --database=microblog --host=10.1.1.1 \
--user=root --password=XXXXXXXX --mode=shard \
--source=replication.json --shard-source=shard.json create-queue
% incline --rdbms=mysql --database=microblog --host=10.1.1.1 \
--user=root --password=XXXXXXXX --mode=shard \
--source=replication.json --shard-source=shard.json create-trigger
The files, `replication.json' and `shard.json' should contain the
definitions shown in the sections above.
RUNNING THE FORWARDER
To transfer modifications between database shards, forwarders should be
run attached to each shard. The example below starts a forwarder process
attached to 10.1.1.1.
% incline --rdbms=mysql --database=microblog --host=10.1.1.1 \
--user=root --password=XXXXXXXX --mode=shard \
--source=replication.json --shard-source=shard.json forward
You should automatically restart the forwarder when it exits (it exits
under certain conditions, for example, when it loses connection to the
attached shard, or when the shard definition is being updated).
SETUP COMPLETE
Now the whole system is up and running. You can try insert / update /
delete the rows in `following' or `tweet' table and see the other tables
updated by incline.
# User:100 starts following user:10100. `Follower' table on 10.1.1.2
# (the shard for user:10100) will be updated
10.1.1.1> INSERT INTO following (user_id,following_id) VALUES \
(100,10100);
10.1.1.2> SELECT * FROM follower WHEER user_id=10100;
+---------+-------------+
| user_id | follower_id |
+---------+-------------+
| 10100 | 100 |
+---------+-------------+
1 row in set (0.00 sec)
# User:10100 tweets. `Timeline' table on 10.1.1.1 will be updated.
10.1.1.2> INSERT INTO tweet (user_id,body) VALUES (10100,'hello');
10.1.1.1> SELECET * FROM timeline WHERE user_id=100;
+---------+---------------+----------+---------------------+
| user_id | tweet_user_id | tweet_id | creation_time |
+---------+---------------+----------+---------------------+
| 100 | 10100 | 1 | 2009-10-05 20:32:07 |
+---------+---------------+----------+---------------------+
1 row in set (0.00 sec)
COMMAND REFERENCE
COMMANDS
create-trigger
Reads the definition files and installs triggers generated onto the
specified database node.
drop-trigger
Reads the definition files and drops the triggers installed from the
specified database node.
print-trigger
Reads the definition files and prints the triggers generated in JSON
format.
create-queue
Reads the definition files and creates queue tables on the specified
database node (only works if --mode is set to either `queue-table'
or `shard).
drop-queue
Reads the definition files and drops the queue tables installed from
the specified database node (only works if --mode is set to either
`queue-table' or `shard').
forward
Reads the definition files and forwards the data from the specified
database node to other nodes (only works if --mode is set to either
`queue-table' or `shard'). The process will stop when connection to
the specified database closes or when the shard definition file is
being updated.
COMMAND OPTIONS
--rdbms=mysql|pgsql
RDBMS being used. Currently supports MySQL (5.0 or above) and
PostgreSQL (8.x?).
--database=db_name
database (schema) name on the database
--host=db_host
hostname of the database. Should be either a hostname or an IP
address (default: 127.0.0.1).
--port=db_port
port number of the database. If ommited, uses the default port
number of the RDBMS.
--user=db_user
username of the database (default: root)
--password=db_passord
password of the databsae (default: none)
--mode=standalone|queue-table|shard
standalone
The mode is for running incline on a single database node. All
updates are reflected synchronously.
queue-table
The mode is for running incline on a single database node. All
updates are queued into the queue tables generated by the
`create-queue' command. The queued updates are applied by the
`forward' command.
shard
The mode is for running incline on multiple database nodes
(shards). Updates that should be applied to the same shard are
applied synchronously. Other updates are pushed into the queue
tables generated by the `create-queue' command. The queued
updates are applied to other nodes by the `forward' command.
--source=replication_def.json
replication definition to be used
--shard-source=shard_def.json
shard definition to be used. Mandatory if --mode is set to `shard'.
--forwarder-log-file=logfile
when set, the `forward' command logs the transactions into the log
file
--version
prints version
--help
prints help
FILE FORMATS
Incline uses two files, replication definition file and shard definition
file. Both of the files use JSON (see RFC 4627 for details) to represent
the structures.
REPLICATION DEFINITION FILE FORMAT
The definition consists of an array. Each element represents a single
replication definition as a hash: between one `destination' table and
more than one `source' tables. The hash may contain following keys.
"destination" : dest_table
name of the destination table
"source" : src_table
"source" : [ src_table_a, src_table_b, ... ]
name of the source table(s)
"pk_columns" : { src_column_a : dest_column_a, ... }
maps columns of source tables(s) to the columns of the destination
table consisting the primary key. Source column should include the
name of the table when using multiple source tables (like:
"src_table_a.column").
"npk_columns" : { src_column_a : dest_column_a, ... }
same as "pk_columns", however defines relations to the
non-primary-key columns of the destination table
"merge" : { src_table_column_a : src_table_column_b, ... }
defines INNER JOIN conditions when using multiple source tables.
"shard-key" : dest_column
when using --mode=shard, defines the column name of the destination
table used as sharding key
SHARD DEFINITION FILE FORMAT
The shard definition file is required only if --mode is set to `shard'.
The file consists of a single JSON hash. Incline recognizes following
keys in the hash.
"algorithm" : "hash-int" | "range-int" (required)
defines the shard algorithm. Incline supports hash-based
partitioning and range-based partitioning of integer columns of
64-bits or smaller.
"num" : number_of_nodes (hash-int only)
defines number of database shards
"nodes" : [ node_def, ... ] (hash-int only)
list of database shards (the number of elements should match the
value of the "num" property)
"map" : { lower-bound : node_def, ... } (range-int only)
list of database shards (keys specify the lower bounds for each
node)
The node definitions in "nodes" or "map" should be a hash or a array of
hashes with following key-value pairs. When using arrays of hashes,
incline will only use the first element of the array. Other elements in
the array may be used by other middlewares such as Pacific, for example
for defining slave database nodes.
"host" : host
hostname or IP address of the database node (required)
"port" : port
port number of the database node (default: uses the default port
number of the RDBMS used)
"username" : db_user
username of the database node (default: "root")
"password" : db_password
password of the database node (default: empty password)
SEE ALSO
Incline & Pacific (in Japanese)
http://www.slideshare.net/kazuho/incline-pacific
A Clever Way to Scale-out a Web Application
http://www.slideshare.net/kazuho/a-clever-way-to-scaleout-a-web-applicat
ion
Kazuho@Cybozu Labs: Intruducing Incline - a synchronization tool for RDB
shards (outdated)
http://developer.cybozu.co.jp/kazuho/2009/07/intruducing-inc.html
AUTHOR
Kazuho Oku <
[email protected]>
LICENSE
The software is licensed under the new BSD license. See "COPYING".
