e3.深入篇-greenplum分布式数据库开发入门到精通41讲课件使用gp mapreduce

1、基于Greenplum Hadoop 分布式平台的大数据解决方案Greenplum Hadoop讲师：迪伦（北风网 )使用GP MapReduce(2) GP MapReduce编程 MapReduce作业故障诊断课程目标GP MapReduce编程定义输出定义OUTPUT是可选的，缺省将最终结果输出到GP MapReduce客户端的标准输出表输出MapReduce生成的结果输出到数据库的表中缺省表如果不存在将会被自动创建如果表存在，需定义一个MODE指定追加(APPEND)输出到该表还是重建(REPLACE)该表如果使用可写外部表，该表必须提交建好，MODE必须为追加(APPEND)- OU

2、TPUT: NAME: gpmr_output TABLE: wordcount_out KEYS: - value MODE: REPLACEGP MapReduce编程文件输出MapReduce生成的结果输出到指定位置的文件中如果文件已经存在，将会返回一个错误- OUTPUT: NAME: gpmr_output FILE: /var/data/mapreduce/wordcount.out定义任务定义TASK是可选的，对于多级MapReduce很有用一个任务定义了完整的端到端INPUT/MAP/REDUCE阶段一个完整的GP MapReduce作业管道一个TASK对象可以被作为其他进一步

3、处理的输入GP MapReduce编程例如，定义2个表输入，一个为documents，另一个为keywords。每个表通过自己的MAP函数document_map和keyword_map来处理。- TASK: NAME: document_prep SOURCE: documents MAP: document_map- TASK: NAME: keyword_prep SOURCE: keywords MAP: keyword_map这些处理阶段的结果作为进一步处理的MapReduce输入- INPUT: NAME: term_join QUERY: | SELECT doc.doc_id,

4、 kw.keyword_id, kw.term, kw.nterms, doc.positions as doc_positions, kw.positions as kw_positions FROM document_prep doc INNER JOIN keyword_prep kw ON (doc.term = kw.term)GP MapReduce编程组装完整的MapReduce定义按照GP MapReduce规范定义了MapReduce所有阶段后，还必须定义一个EXECUTE部分以指定最终的INPUT/MAP/REDUCE阶段EXECUTE: - RUN: SOURCE: in

5、put_or_task_name TARGET: output_name MAP: map_function_name REDUCE: reduce_function_nameGP MapReduce编程提交MapReduce作业执行通过gpmapreduce客户端程序提交该程序到GP提供连接信息，如数据库、用户名、主机名和端口等gpmapreduce -h gpmasterhost -p 5432 -f my_gpmr_spec.yml mydatabase在数据库中创建语言在执行Map和Reduce函数时，GP MapReduce会使用过程语言将这些函数构建到数据库中GP需要提前在数据库中

6、创建这些语言使用CREATE LANGUAGE命令在数据库中为要执行的MapReduce作业创建语言必须由超级用户创建$ psql c CREATE LANGUAGE plperl;创建Python过程语言：$ psql c CREATE LANGUAGE plpython;GP MapReduce编程安装库文件(自定义C函数)函数定义的库文件必须安装到GP所有主机相同路径例如：在MapReduce的YAML定义中存在一个C函数的定义：DEFINE: - TRANSITION: NAME: int4_accum PARAMETERS: state int8, value int4 RETURN

7、S: state int8 LANGUAGE: C LIBRARY: $libdir/gpmr.so FUNCTION: int4_accumLIBRARY指定的$libdir/gpmr.so文件必须在所有主机的$GPHOME/lib目录下存在MapReduce作业故障诊断语言不存在症状ERROR: language pl* does not existHINT: Use CREATE LANGUAGE to load the language into the database.解释在执行Map和Reduce函数时，GP MapReduce会使用过程语言将这些函数构建到数据库中。GP需提前创

8、建好这些语言解决方案创建相应的语言，如：psql database_name c CREATE LANGUAGE plperl;MapReduce作业故障诊断通用Python迭代器错误症状ERROR: plpython: function “function_name” error fetching next item from iteratorDETAIL: Errno 2 No such file or directory: /tmp/file/doesnt/exist (, line 39)解释当Python的迭代函数中存在错误的时候，就会返回该错误；这通常标识着你的Python程序存在

9、bug解决方案最简单的调试办法是在函数体外包装一个错误捕获结构。例如：FUNCTION: | try: . user code . except Exception, e: plpy.warning(my function name: + str(e)MapReduce作业故障诊断函数定义使用了错误的模式症状ERROR: set-returning PERL function must return reference to array or use return_next-ERROR: composite-returning PERL function must return referen

10、ce to hash-ERROR: returned sequences length must be same as tuples length-ERROR: no attribute named keyHINT: to return null in specific column, let returned object to have attribute named after column with value None-MapReduce作业故障诊断解释在GP MapReduce函数中有两种主要模式：MODE: SINGLE 每行接收返回单行MODE: MULTI 每行接收可能返回0

11、N的任意行Transition函数和Consolidate函数仅支持SINGLE模式。因为他们属于有限状态函数，且必须返回下一个状态Map函数和Finalize函数支持两种模式且缺省为Multi模式解决方案在SINGLE模式的函数中必须使用return方法来返回单行数据在MULTI模式中，最好作为通用函数来编写，在Python中使用yield，在Perl中使用return_nextMapReduce作业故障诊断例如- MAP: NAME: perl_single MODE: SINGLE PARAMETERS: key text, value text RETURNS: key text, v

12、alue text LANGUAGE: perl FUNCTION: | my ($key, $value) = _; return key = $key, value = $valuereturn undef MapReduce作业故障诊断- MAP: NAME: perl_multi MODE: MULTI PARAMETERS: key text, value text RETURNS: key text, value text LANGUAGE: perl FUNCTION: | my ($key, $value) = _; for my $i (0.10) return_next k

13、ey = $key, value = $value return undef MapReduce作业故障诊断- MAP: NAME: python_single MODE: SINGLE PARAMETERS: key text, value text RETURNS: key text, value text LANGUAGE: python FUNCTION: | return key: key, value: value - MAP: NAME: python_multi MODE: MULTI PARAMETERS: key text, value text RETURNS: key text, value text LANGUAGE: python FUNCTION: | try: for i in range(0,10): yield key: key, value: value except Exception, e: plpy.warni