MapReduce + Oracle = Tablefunctions
我们在OpenWorld大会做的其中一件事，是漂亮的展示了如何在通用的Oracle数据库之上实现MapReduce系统。这里基于在这个博客上，显示了很好的实施tablefunctions和映射器等.
但后来我们想，为什么不经过tablefunction代码和MapReduce范例一种映射来告诉大家在Oracle中构建存在，并且如何利用Oracle创建一个数据处理/分析管道...所以这里是一些我们在OpenWorld大会中正在使用的代码.
承上启下，首先我们讨论的标题，高亮有趣的片段与代码注释，然后我们讨论的主体和实际（简单）映射器，减速机代码。该意见是很有希望使这个东西不言自明的...
Scenario
我们在这里做的相当简单。创建一个简单的表，表里有一些记录和循环。减速器是做一个聚集。步骤如下：
CREATE TABLE sls (salesman VARCHAR2(30), quantity number)
/
INSERT INTO sls VALUES('Tom', 100);
INSERT INTO sls VALUES('Chu', 200);
INSERT INTO sls VALUES('Tom', 300);
INSERT INTO sls VALUES('Mike', 100);
INSERT INTO sls VALUES('Scott', 300);
INSERT INTO sls VALUES('Tom', 250);
INSERT INTO sls VALUES('Scott', 100);
commit;
/
Header
create or replace package oracle_map_reduce
is
-- The types we define here is similar to the input files
-- and output files that are used in MR code and are used to
-- store data while we run the actual package.
-- The big advantage is that we do not need to write to disk for
-- intermediate results.
type sales_t is table of sls%rowtype;
type sale_cur_t is ref cursor return sls%rowtype;
type sale_rec_t is record (name varchar2(30), total number);
type total_sales_t is table of sale_rec_t;
-- Next we define the funtions that do the work and make them known
-- to the outside world
-- Note that both mapper and reducer are tablefunctions!
-- Both mapper and reducer are pipelined and executable in parallel
-- the parallel degree is driven from the database side and is not
-- scheduled by the actual program
function mapper(inp_cur in sys_refcursor) return sales_t
pipelined parallel_enable (partition inp_cur by any);
-- the pipelined keyword tells the caller that this function acts as
-- a row source
--
-- parallel_enable indicates that this function can be executed in parallel
-- by the parallel query framework.
function reducer(in_cur in sale_cur_t) return total_sales_t
pipelined parallel_enable (partition in_cur by hash(salesman))
-- Finally we can cluster the results so that similar rows are chunked
-- together when used (note this does not drive distribution over the
-- parallel slaves, which is done by the partition clause shown in the mapper
-- and reducers)
cluster in_cur by (salesman);
end;
/
-- The body of the package has the mapper and the reducer code
-- The header as is shown here by itself defines the signature of
-- the package and declares types and variables to be used in the
-- package.
Body
create or replace package oracle_map_reduce
is
type sales_t is table of sls%rowtype;
type sale_cur_t is ref cursor return sls%rowtype;
type sale_rec_t is record (name varchar2(30), total number);
type total_sales_t is table of sale_rec_t;
function mapper(inp_cur in sys_refcursor) return sales_t
pipelined parallel_enable (partition inp_cur by any);
function reducer(in_cur in sale_cur_t) return total_sales_t
pipelined parallel_enable (partition in_cur by hash(salesman))
cluster in_cur by (salesman);
end;
/
-- The upper part is the header the following part if the body
-- Note the difference in the create statement below as compared
-- to the header
create or replace package body oracle_map_reduce
is
function mapper(inp_cur in sys_refcursor) return sales_t
pipelined parallel_enable (partition inp_cur by any)
is
sales_rec sls%ROWTYPE;
-- construct a record to hold an entire row from the SLS table
begin
-- First loop over all records in the table
loop
fetch inp_cur into sales_rec;
exit when inp_cur%notfound;
-- Place the found records from SLS into the variable
-- end the loop when there are no more rows to loop over
pipe row (sales_rec);
-- by using pipe row here we are giving back rows in streaming
-- fashion as you would expect from a table
-- this in combination with pipelined in the definition allows
-- the pipelining (e.g. giving data as it comes on board) of
-- a table function
end loop;
return;
-- Return is a mandatory piece that allows the consumer of data (our reducer
-- in this case)
-- to ensure all data has been sent. After return the rowsource is exhausted
-- and no more data comes from this function.
end mapper;
-- The above mapper does in effect nothing other than streaming data
-- partitioned
-- over to the next step. In MR the stream would be written to a file and then -- redistributed to the reducers
-- The reducer below computes and emits the sales figures
function reducer(in_cur in sale_cur_t) return total_sales_t
pipelined parallel_enable (partition in_cur by hash(salesman))
-- The partition by clause indicates that all instances of a particular
-- salesman must be sent to one instances of the reducer function
cluster in_cur by (salesman)
-- The cluster by clause tells the parallel query framework to cluster
-- all instances of a particular salesman together.
IS
sale_rec sls%ROWTYPE;
total_sale_rec sale_rec_t;
-- two containers are created, one as input the other as output
begin
total_sale_rec.total := 0;
total_sale_rec.name := NULL;
-- reset the values to initial values
loop
fetch in_cur into sale_rec;
exit when in_cur%notfound;
-- some if then logic to ensure we pipe a row once all is processed
if (total_sale_rec.name is null) then
-- The first instance is arriving, set the salesman value to that
-- input value
-- update 0 plus the incoming value for total
total_sale_rec.name := sale_rec.salesman;
total_sale_rec.total := total_sale_rec.total +
sale_rec.quantity;
elsif ( total_sale_rec.name <> sale_rec.salesman) then
-- We now switch sales man, and are done with the first
-- salesman (as rows are partitioned and clustered)
-- First pipe out the result of the previous salesman we
-- processed
-- then update the information to work on this new salesman
pipe row (total_sale_rec);
total_sale_rec.name := sale_rec.salesman;
total_sale_rec.total := sale_rec.quantity;
else
-- We get here when we work on the same salesman and just add
-- the totals, the move on to the next record
total_sale_rec.total := total_sale_rec.total +
sale_rec.quantity;
end if;
end loop;
-- The next piece of code ensures that any remaining rows that
-- have not been piped out
-- are piped out to the consumer. If there is a single salesman,
-- he is only piped out
-- in this piece of logic as we (in the above example code) only
-- pipe out upon a change
-- of salesman
if total_sale_rec.total<> 0 then
pipe row (total_sale_rec);
end if;
return;
-- Again, we are now done and have piped all rows to our consumer
end reducer;
end;
/
在一个SQL 查询中使用它
花了一点时间，但是一旦你看到查询，你就可以知道如何建立一系列预定义的程序，然后你可以实现
串在一起获得一组结果集。
select *
from table(oracle_map_reduce.reducer(cursor(
select * from table(oracle_map_reduce.mapper(cursor(
select * from sls))) map_result)));
所有的逻辑管道数据都到下一个消费者，并且所有都是并行运行的。这使得它适合类似重型数据库ETL（我们首先为了它发明的）的任何东西，并且任何需要应用大量逻辑到记录的东西（像分析处理）。