学步园

背景：
　　现有一个数据表task，目前已有数据位4000万左右，因为表太大，影响到数据库的操作性能，所以考虑对该task表中的数据进行重新分片。
如果是在数据库设计的时候就考虑到这样的问题，可以采用postgresql的分区表，通过表继承以及创建一些trigger或者rules来实现这样的要求。但目前task表中已有大量的数据，所以直接多task表采用继承的方式来实现分片已经不太现实，另外insert到task表中时date_create已有值，但最终的表分区应该按照date_start，date_start只有在后期才会更新，在date_create时date_start为空；上层的应用已经成型，不可能更改代码。所以综合考虑下来采用传统的分区技术已经不太现实或者代价太高，无奈之下只能自己动手写一写plpgsql函数，通过crontab制定定时任务来对task表中的数据进行分块。
具体代码如下：

--创建相对应的表
CREATE OR REPLACE FUNCTION do_create_partition(table_name TEXT)
RETURNS TEXT AS 
$BODY$
DECLARE
BEGIN
    RAISE NOTICE 'Begin to create patition ...';

    RAISE NOTICE 'partition table name = %', table_name;
--创建分区
    /*
    EXECUTE 'CREATE TABLE '
        || table_name
        || '() INHERITS (task);';
*/
EXECUTE  'CREATE TABLE '
|| table_name
|| '('
|| 'task_id integer NOT NULL,'
|| 'task_name text,'
|| 'task_type text,'
|| 'priority integer,'
|| 'expires integer,'
|| 'countdown integer,'
|| 'tube_name text,'
|| 'kwargs text,'
|| 'retries integer,'
|| 'status text,'
|| 'result text,'
|| 'date_start timestamp without time zone,'
|| 'date_done timestamp without time zone,'
|| 'traceback text,'
|| 'date_create timestamp without time zone,'
|| 'worker_id integer'
|| ')';
--创建主键
    EXECUTE  'ALTER TABLE '
|| table_name
|| ' ADD CONSTRAINT '
|| table_name
|| '_task_id '
' PRIMARY KEY(task_id);';

    RAISE NOTICE '% created"', table_name;

RETURN 'DONE';
END
$BODY$ 
LANGUAGE 'plpgsql'

--创建所需的分区表
CREATE OR REPLACE FUNCTION create_partition_table()
RETURNS TEXT AS 
$BODY$
DECLARE
    p_table_name TEXT;
    now_dt TIMESTAMP;
    table_exist int;
BEGIN
    RAISE NOTICE 'Begin to create partition table ...';
--INSERT INTO task_20111018 VALUES (r.*)
    -- construct the table name
    --table_name := 'task_' || to_char(r.date_start, 'YYYYMMDD');
    --获取当前时间
    SELECT CURRENT_TIMESTAMP INTO now_dt;
    p_table_name := 'task_' || to_char(now_dt, 'YYYYMMDD');
    RAISE NOTICE 'partition table = %', p_table_name;
--检测所需的分区表是否存在
    SELECT count(*) INTO table_exist FROM pg_tables WHERE schemaname='public' and  tablename=p_table_name;
IF table_exist = 0 THEN
--创建当前分区表
        RAISE NOTICE 'CREATE current partition table...';
        PERFORM do_create_partition(p_table_name);
ELSE
--检测创建第二天的分区表
        now_dt := now_dt + interval '1 day';
        p_table_name := 'task_' || to_char(now_dt, 'YYYYMMDD');
        RAISE NOTICE 'partition table = %', p_table_name;
--检测所需的分区表是否存在
        SELECT count(*) INTO table_exist FROM pg_tables WHERE schemaname='public' and  tablename=p_table_name;
IF table_exist = 0 THEN
--创建当前分区表
            RAISE NOTICE 'CREATE next partition table...';
            PERFORM do_create_partition(p_table_name);
END IF;
END IF;
--INSERT INTO table_name VALUES (r.*);
    RAISE NOTICE 'DO..."';

RETURN 'DONE';
END
$BODY$ 
LANGUAGE 'plpgsql'


--将task中的已有数据根据date_start转移到相应的分区表中
CREATE OR REPLACE FUNCTION slice_data()
RETURNS TEXT AS 
$BODY$
DECLARE
    r task%rowtype;
    table_name TEXT;
    table_exist int;
    date_format TEXT;
BEGIN
    RAISE NOTICE 'Begin to slice data ...';
--将date_start在今天之前的相关记录进行转储
    FOR r IN SELECT * FROM task WHERE status in ('SUCCESS', 'FAILURE' ) and date_start is not NULL and date_start < current_date::timestamp --limit 1000000
    LOOP 
--INSERT INTO task_20111018 VALUES (r.*)
        -- construct the table name
        table_name := 'task_' || to_char(r.date_start, 'YYYYMMDD');
--table_name = 'task_20111018';
        RAISE NOTICE 'table_name = %', table_name;
--INSERT INTO table_name VALUES (r.*);
        --检测所需的分区表是否存在
        SELECT count(*) INTO table_exist FROM pg_tables WHERE schemaname='public' and  tablename=table_name;
IF table_exist = 0 THEN
--创建所需分区
            PERFORM do_create_partition(table_name);
END IF;

        date_format := 'YYYY-MM-DD HH24:MI:SS';
        RAISE NOTICE 'do insert [task_id = %] ...  ', r.task_id;
EXECUTE  'INSERT INTO '
|| table_name
|| '(task_id, task_name, task_type, priority, expires, countdown, tube_name, kwargs, retries, status, result, date_start, date_done, traceback, date_create, worker_id)'
|| ' VALUES ('
|| r.task_id
|| ','
|| COALESCE(quote_literal(r.task_name), 'DEFAULT')
|| ','
|| COALESCE(quote_literal(r.task_type), 'DEFAULT')
|| ','
|| r.priority
|| ','
|| r.expires
|| ','
|| r.countdown
|| ','
|| COALESCE(quote_literal(r.tube_name), 'DEFAULT')
|| ','
|| COALESCE(quote_literal(r.kwargs), 'DEFAULT')
|| ','
|| r.retries
|| ','
|| COALESCE(quote_literal(r.status), 'DEFAULT')
|| ','
|| COALESCE(quote_literal(r.result), 'DEFAULT')
|| ','
|| COALESCE(quote_literal(to_char(r.date_start, date_format)), 'DEFAULT')
|| ','
|| COALESCE(quote_literal(to_char(r.date_done, date_format)), 'DEFAULT')
|| ','
|| COALESCE(quote_literal(r.traceback), 'DEFAULT')
|| ','
|| COALESCE(quote_literal(to_char(r.date_create, date_format)), 'DEFAULT')
|| ','
|| COALESCE(r.worker_id, 0)
|| ')';

DELETE FROM task WHERE task_id = r.task_id;
        RAISE NOTICE 'delete old record';
END LOOP;
RETURN 'DONE';
END
$BODY$ 
LANGUAGE 'plpgsql'

可以在crontab中添加如下任务来实现对task已有数据的分块操作
0 2 * * * /opt/pgsql9.1/bin/psql -d taskmanager -U postgres -c "select * slice_data();"