Tarantoolは、データウェアハウスのタプルキーです。 すべてのデータとインデックスはRAMに保存されます。 値はタプルを構成し、次にタプル 、タプル-スペース、次にスペース 、スペース-データモデルを構成します。 3種類のデータがサポートされています：32ビット符号なし整数、64ビット符号なし整数、バイナリ文字列、それぞれNUM 、 NUM64およびSTR 。 スペースの場合、プライマリインデックスのタイプと構造を定義する必要があります。たとえば、フィールド1,2のHASH（1はNUM、2はNUM64）。 セカンダリインデックスは、プライマリインデックスと同じ方法で指定されます。 DML操作はタプルレベルでアトミックであり、プライマリインデックスでのみ実行されます。 複数の操作をアトミックに実行するには、組み込み言語Luaを使用する必要があります。 データの安全性は、現在の状態のスナップショット 、 スナップショット 、バイナリログ、 xlogの順に保存することで確保されます 。 タプルの保管には、 スラブが使用されます。
以下の例では、javaコネクタが使用されています。詳細については、 dgreenru.github.com / tarantool- javaを参照してください。 執筆時点での最新の安定バージョンは0.1.2です。 以下では、他のリポジトリとデータを転送および同期できる追加機能の使用例を検討します。
MySQLテーブルをTarantool Boxに移行する例：
mysql> desc user; +------------+--------------+------+-----+-------------------+----------------+ | Field | Type | Null | Key | Default | Extra | +------------+--------------+------+-----+-------------------+----------------+ | id | int(11) | NO | PRI | NULL | auto_increment | | username | varchar(255) | NO | UNI | NULL | | | email | varchar(255) | NO | UNI | NULL | | | enabled | tinyint(1) | NO | | 1 | | | registered | timestamp | NO | | CURRENT_TIMESTAMP | | +------------+--------------+------+-----+-------------------+----------------+ 5 rows in set
プライマリインデックスIDと2つのセカンダリユニークインデックスユーザー名とメール。 デフォルトでは許容されない場所のうち、 auto_incrementとtimestampは区別できます。 最初の場合はbox.auto_incrementストアドプロシージャを使用でき、2番目の場合はyyyyMMddhhmmss形式または秒でデータを保存できます。 ユーザーテーブルが十分に小さい場合は、mysqlからデータを読み取ってTarantool Boxに貼り付けることができます。このタスクは停止しませんが、テーブルが非常に大きい場合、つまりテーブルが非常に大きい場合の対処方法を説明します。 各レコードは小さいですが、多くのレコードが含まれています。 まず、できればサーバーリソースの多くを占有しないで、便利な形式でデータをアップロードする必要があります。
mysql> select * into outfile '/tmp/user' from user; Query OK, 73890541 rows affected $ head -1 /tmp/user 1 username email@domain.tld 1 2012-10-14 01:27:05
ファイルを目的のサーバーまたはローカルコンピューターにコピーすることにより、その処理とTarantool Box形式への変換に進むことができます。 以下の例では、簡単にするために、エスケープシーケンスは考慮されていません。 テーブルにタブ、改行、キャリッジリターン、バックスラッシュ、またはNULL値を含むフィールドがある場合、それらの処理を自分で追加する必要があります。
BufferedReader reader = new BufferedReader(new InputStreamReader(new GZIPInputStream(new FileInputStream("/tmp/user.gz")), "utf-8")); SnapshotWriter writer = new SnapshotWriter(new FileOutputStream("/tmp/user.snap").getChannel()); String line = null; DateFormat indf = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss"); DateFormat outdf = new SimpleDateFormat("yyyyMMddhhmmss"); Pattern pattern = Pattern.compile("\t"); while ((line = reader.readLine()) != null) { try { String[] values = pattern.split(line); if (values.length == 5) { Integer id = Integer.parseInt(values[0]); String username = values[1]; String email = values[2]; byte[] enabled = { Byte.valueOf(values[3]) }; Long registered = Long.parseLong(outdf.format(indf.parse(values[4]))); Tuple tuple = new Tuple(5).setInt(0, id).setString(1, username, "UTF-8") .setString(2, email, "UTF-8").setBytes(3, enabled).setLong(4, registered); writer.writeRow(0, tuple); } else { System.err.println("Line should be splited in 5 parts, but has " + values.length + " for " + line); } } catch (Exception e) { System.err.println("Can't parse line " + line); e.printStackTrace(); } } writer.close(); reader.close();
その結果、ファイルがあります
$ ls -sh /tmp/user.snap 16.1G /tmp/user.snap
次に、それに応じてスペース0を構成する必要があります。
# slab . # slab, # 2 . # 24 . slab_alloc_arena = 24 # xlog . rows_per_wal = 500000 # space 0 space[0].enabled = 1 # id. box.auto_increment TREE. space[0].index[0].type = "TREE" space[0].index[0].unique = 1 space[0].index[0].key_field[0].fieldno = 0 space[0].index[0].key_field[0].type = "NUM" #username space[0].index[1].type = "HASH" space[0].index[1].unique = 1 space[0].index[1].key_field[0].fieldno = 1 space[0].index[1].key_field[0].type = "STR" #password space[0].index[2].type = "HASH" space[0].index[2].unique = 1 space[0].index[2].key_field[0].fieldno = 2 space[0].index[2].key_field[0].type = "STR"
次に、構成ファイルのwork_dirフォルダーにある00000000000000000001.snapを、作成したファイルに置き換える必要があります。
$ mv /tmp/user.snap /var/lib/tarantool/00000000000000000001.snap
そして、サーバーを起動してみてください
$ tarantool_box --background $ ps -C tarantool_box -o pid=,cmd= 8853 tarantool_box: primary pri: 33013 sec: 33014 adm: 33015
また、tarantool.logファイルを調べて、正常に起動すると、以下のような行で終了します。エラーが発生した場合は、すぐに理由が表示されます。
1350504007.249 7127 1/sched _ I> Space 0: done 1350504007.249 7127 101/33013/primary _ I> bound to port 33013 1350504007.249 7127 101/33013/primary _ I> I am primary 1350504007.249 7127 102/33014/secondary _ I> bound to port 33014 1350504007.250 7127 103/33015/admin _ I> bound to port 33015 1350504007.251 7127 1/sched _ C> log level 4 1350504007.251 7127 1/sched _ C> entering event loop
さらに、データ挿入の正確性は簡単な方法で確認できます。
$ tarantool -a 127.0.0.1 127.0.0.1> select * from t0 where k0 = 1 Select OK, 1 rows affected [1, 'username', 'email@domain.tld', '\x01', '\x21\x8b\xe4\xc9\x4c\x12'] 127.0.0.1> select * from t0 where k1 = 'username' Select OK, 1 rows affected [1, 'username', 'email@domain.tld', '\x01', '\x21\x8b\xe4\xc9\x4c\x12'] 127.0.0.1> select * from t0 where k2 = 'email@domain.tld' Select OK, 1 rows affected [1, 'username', 'email@domain.tld', '\x01', '\x21\x8b\xe4\xc9\x4c\x12']
つまり configで指定された3つのキーのデータの場所を確認しました。 次に、システム内のプロセスが消費したメモリ量と、Tarantool Boxコンソールのshow slabコマンドのレポートを確認できます。
Tarantool Boxが起動しました。ここで、一部のクエリが独自の目的でデータを使用する場合に備えて、データをバックアップし、MySQLテーブルを最新の状態に保つ必要があります。 ReplicationClientクラスを使用して整理するのは非常に簡単です。 本格的なスレーブサーバーを使用せずにほぼ完全なxlogバックアップを作成し、追加のリソースと時間をかけずにMySQLでテーブルを更新することができます。 構成でreplication_portを指定して、複製を可能にすることを忘れないでください。 以下に説明するクラスでは、サーバーから受信したすべてのログを5万件のレコードの長さのファイルに保存します。 操作アルゴリズムは非常に簡単です。
1.既存のログを検索する
2.最大lsnの決定
3.レプリケーションポートへの接続
4.受信したデータをファイルに変換します
このコードにはMySQL更新ロジックがありませんが、main関数のループをわずかに変更することで簡単に実装できます。 難しい場所は、受信データをバイナリログに書き込み、xlog形式に拡張するコードを使用したReplicationClientクラスの拡張です。 この場所で特に立ち止まることはできません。なぜなら この例は、使用例よりも実際のアプリケーションのプレフィックスである可能性が高くなります。
public class Backup { protected DecimalFormat xlogNameFormat = new DecimalFormat("00000000000000000000"); protected String folder; protected FileChannel xlogChannel; protected int row; protected int limit = 50000; protected long lsn = 0L; protected ReplicationClient client; protected XLogWriter writer; public void setLimit(int limit) { this.limit = limit; } public Backup(String folder, String host, int port) throws IOException { this.folder = folder; } protected void getLatestLSN(String folder) throws IOException, FileNotFoundException { final File backupFolder = new File(folder); String[] xlogs = backupFolder.list(new FilenameFilter() { @Override public boolean accept(File dir, String name) { return name.endsWith(".xlog"); } }); boolean hasLogs = xlogs != null && xlogs.length > 0; if (hasLogs) { Arrays.sort(xlogs); XLogReader reader = new XLogReader(new FileInputStream(folder + "/" + xlogs[xlogs.length - 1]).getChannel()); XLogEntry xlogEntry = null; while ((xlogEntry = reader.nextEntry()) != null) { lsn = xlogEntry.header.lsn; } reader.close(); } } public void start() throws IOException { getLatestLSN(folder); System.out.println("Planning to start from lsn: " + lsn); Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() { @Override public void run() { try { synchronized (this) { close(); } } catch (IOException e) { throw new IllegalStateException("Can't close xlog", e); } } })); final ByteBuffer rowStartMarker = ByteBuffer.allocate(4).order(ByteOrder.LITTLE_ENDIAN).putInt(Const.ROW_START_MARKER); client = new ReplicationClient(SocketChannel.open(new InetSocketAddress("127.0.0.1", 33016)), lsn + 1L) { @Override protected ByteBuffer readBody(Header header) throws IOException { if (Backup.this.xlogChannel == null) { Backup.this.xlogChannel = nextFile(folder); } ByteBuffer body = super.readBody(header); this.header.flip(); rowStartMarker.flip(); synchronized (Backup.this) { while (rowStartMarker.hasRemaining()) Backup.this.xlogChannel.write(rowStartMarker); while (this.header.hasRemaining()) Backup.this.xlogChannel.write(this.header); while (body.hasRemaining()) Backup.this.xlogChannel.write(body); Backup.this.xlogChannel.force(false); body.flip(); } return body; } }; } public XLogEntry nextEntry() throws IOException { XLogEntry entry = client.nextEntry(); lsn = entry.header.lsn; if (++row >= limit) { close(); xlogChannel = nextFile(folder); row = 0; } return entry; } protected FileChannel nextFile(String folder) throws IOException { String fileName = folder + "/" + xlogNameFormat.format(lsn + 1L) + ".xlog"; new File(fileName).createNewFile(); FileChannel channel = new FileOutputStream(fileName, true).getChannel(); writer = new XLogWriter(channel); return channel; } public void close() throws IOException { if (writer != null) { writer.close(); } } public static void main(String[] args) throws IOException { final Backup backup = new Backup("/home/dgreen/backup", "localhost", 33016); backup.start(); XLogEntry entry = null; while ((entry = backup.nextEntry()) != null) { StringBuilder pk = new StringBuilder(); for (int i = 0; i < entry.tuple.size(); i++) { if (pk.length() > 0) { pk.append(" - "); } switch (entry.tuple.getBytes(i).length) { case 4: pk.append(String.valueOf(entry.tuple.getInt(i))); break; case 8: pk.append(String.valueOf(entry.tuple.getLong(i))); break; default: pk.append(entry.tuple.getString(i, "UTF-8")); } } switch (entry.op) { case Update.OP_CODE: System.out.println("Got update on #" + pk.toString()); break; case Insert.OP_CODE: System.out.println("Got insert " + pk.toString()); break; case Delete.OP_CODE: System.out.println("Got delete of #" + pk.toString()); break; default: System.out.println("Got unknown op " + entry.op + " " + pk.toString()); break; } } } }
また、xlogファイルを操作する機能を使用する場合 、 XLogReaderクラスとXLogWriterクラスを使用して誤ってタプルを削除したりスペースを完全に削除したりしても、データを失うことはほとんど不可能であることに注意してください。xlogを簡単に編集できます。
基本的にはこれですべてです。コネクタについて詳しくはdgreenru.github.com/tarantool-javaで学習できることを思い出してください。使用されているサンプルのソースコードはgithubリポジトリにあります。