[Translate] 从SQL数据库迁移到Riak

例子

关系数据库: 创建表

CREATE TABLE posts (
    id SERIAL PRIMARY KEY,
    author VARCHAR(30) NOT NULL,
    title VARCHAR(50) NOT NULL,
    body TEXT NOT NULL,
    created DATE NOT NULL
);

关系数据库: 查询

SELECT * FROM posts WHERE id = 99;

关系数据库: 查询结果

 id |   author   |             title              |            body            |  created
----+------------+--------------------------------+----------------------------+------------
 99 | John Daily | Riak Development Anti-Patterns | Writing an application ... | 2014-01-07

基本转换和存储过程如下

每一个表行被转换为一个JSON对象,包含除id字段外的所有其他字段.
id字段不会存储在Riak中的JSON对象中. 它不会作为对象的键. 而是把标题作为键, 30字符长度限制, 小写, 并且用-作为单词之间的连字符. 比如上面的例子把riak-development-anti-patterns作为键.
所有产生自posts表的JSON对象存储在名为posts的单个Riak桶当中.
键存在在Riak set中, 需要的时候, 使所有对象可以被一次查询出来.

把Table转换为List

下面, 我们使用psycopg2库来把表转换为列表

import psycopg2
connection = psycopg2.connection('dbname=blog_db')
cursor = connection.cursor()

用该游标(cursor)执行SELECT * FROM posts查询,然后使用fetchall函数从游标当中获取信息:

cursor.execute('SELECT * FROM posts')
table = cursor.fetchall()

table对象由一个Python元组列表构成, 如下:

[(1, 'John Doe', 'Post 1 title', 'Post body ...', datetime.date(2014, 1, 1)),
 (2, 'Jane Doe', 'Post 2 title', 'Post body ...', datetime.date(2014, 1, 2)),
 # more posts in the list
]

转换行为JSON对象

上面的代码把posts表中的每一行数据转换为包含5个元素的元组, 为了更好的进行转换, 我们把元组转换为能够在Riak中更好的处理的Python字典数据类型, 官方的Riak Python客户端能够自动地把Python字段转换为能够在Riak中存储的JSON对象. 一旦我们有了一个字典列表, 就可以直接在Riak中存储这些字典类型的数据.

下面通过Python代码把数据库中的数据转换为JSON格式:

import datetime

def convert_row_to_dict(row):
    return {
        'author': row[1],
        'title': row[2],
        'body': row[3],
        'created': row[4].strftime('%m-%d-%Y')
    }

产生如下的JSON数据:

{
  'author': 'John Daily',
  'title': 'Riak Development Anti-Patterns',
  'body': 'Writing an application ...',
  'created': '01-07-2014'
}

存储行对象

现在使用store_row_in_riak把posts表中的数据存储到Riak中:

用博客的标题来构造键, 获取签名30字符, 字符全部转换为小写, 并用-替换空格.
每行将被转换为一个正确的Riak对象,并存储

下面是这个函数的实现:

bucket = client.bucket('posts')

def store_row_in_riak(row):
    key = row[2][0:29].lower().replace(' ', '-') 
    obj = RiakObject(client, bucket, key)
    obj.content_type = 'application/json'
    obj.data = convert_row_to_dict(row)
    obj.store()

上面我们在一个Riak set存储所有对象的键, 以帮助我们在将来能够查询这些对象. 我们修改上的store_row_in_riak函数添加每一个键到一个set(集合):

from riak.datatypes import Set

objects_bucket = client.bucket('posts')
key_set = Set(client.bucket_type('sets').bucket('key_sets'), 'posts')

def store_row_in_riak(row):
    key = row[0]
    obj = RiakObject(client, bucket, key)
    obj.content_type = 'application/json'
    obj.data = convert_row_to_dict(row)
    obj.store()

现在,我们编写一个迭代器用于存储所有的行:

# Using our "table" object from above:

for row in table:
    store_row_in_riak(row)

一旦所有对象存储到Riak, 我们可以执行正常的键/值操作来一个一个地获取. 下面是一个使用Riak HTTP API的例子:

curl http://localhost:8098/buckets/posts/keys/99

这将返回一个博客(Blog)对象:

{
  "author": "John Daily",
  "title": "Riak Development Anti-Patterns",
  "body": "Writing an application ...",
  "created": "01-07-2014"
}

需要的时候, 我们也可以一次获取所有这些对象. 前面我们在Raik set中存储了所有对象的键. 我们可以编写一个函数从set中获取所有对象的key, 以及和这些key相关的所有对象:

from riak.datatypes import Set

set_bucket = client.bucket_type('sets').bucket('key_sets')
posts_bucket = client.bucket('posts')

def fetch_all_objects(table_name):
    keys = Set(client, bucket, table_name)
    for key in keys:
        return posts_bucket.get(key)

fetch_all_objects('posts')

这回返回之前存储的Python字典的完整列表.

使用二级索引

添加一个关键字字段(keywords), 来描述一篇博客的关键字列表:

CREATE TABLE posts (
    id SERIAL PRIMARY KEY,
    author VARCHAR(30) NOT NULL,
    title VARCHAR(50) NOT NULL,
    body TEXT NOT NULL,
    created DATE NOT NULL,
    keywords TEXT[] NOT NULL
);

插入数据:

INSERT INTO posts (author, title, body, created, keywords) VALUES
    ('Basho', 'Moving from MySQL to Riak', 'Traditional database architectures...',
    current_date, '{"mysql","riak","migration","rdbms"}');

为没一个博客的关键字列表添加一个二级索引. 下面我们编写一个函数对每一个关键字附加一个二级索引.

def add_keyword_2i_to_object(obj, keywords):
    for keywork in keywords:
        obj.add_index('keywords_bin', keyword)

然后我们可以插入该函数到store_row_in_riak函数中:

bucket = client.bucket('posts')

def store_row_in_riak(row):
    obj = RiakObject(client, bucket, row[0])
    obj.content_type = 'application/json'
    obj.data = convert_row_to_dict(row)
    add_keyword_2i_to_object(obj, row[5])
    obj.store()

现在我们可以基于一篇博客的关键字来查询:

bucket = client.bucket('posts')

def fetch_posts_by_keyword(keyword):
    for key in bucket.get_index('keywords_bin', keyword):
        return bucket.get(key)

该函数会返回所有包含特定关键字的博客列表.