xml in hadoop ETL with pig summary

lx86

　　项目中需要把source为xml的文件通过flume放置到hdfs,然后通过MR导入到vertica中去,我之前做过简单的
　　尝试,是通过pig的piggybank的xmlloader然后Regex_extract来提取结点属性做的,但问题是我之前只取了一
　　层结点的属性,没有把不同层次结点关联起来,这有三四层,结构比较复杂,我需要重新整理思路.
　　这种方式很可能走不通,因为piggybank里面regex_extract的正则和传统的正则还是有些异同的.常常会
　　因为正则写的不合适经常返回空元组.
　　我是一个c# guy,又不会用纯java写MR,所以就进一步搜索了google.查找相关资料.
　　1.把XML先转成avro的形式,然后再使用pig 进行转换,可以减少CPU的利用率和IO的情况,具体可以参见slides:
　　http://www.slideshare.net/Hadoop_Summit/bose-june26-405pmroom230cv3-24148869
　　不清楚Avro,所以个人尚未尝试过此种处理方式.
　　2.使用StreamingXMLLoader,来处理复杂的巨大的xml文件.
　　可以参见这篇文章How to feed XML to your Pig
　　http://www.tuicool.com/articles/vEJbUj
　　The class is in a third party project :Mortar.
　　这是一个第三方的工程,我也不清楚,所以也放弃了.
　　3.使用Mahout的相关类.不清楚,未使用过.但搜索出来说mahout中相关的文档格式处理可以做这件事.
　　4.使用pig的piggybank中xmlloader,但xmlloader用起来虽然简单,但其思路也简单,就是传入标签,然后获取
　　此标签下的所有原生的XML,这对于此标签下有更多的复杂元素时,进一步处理起来还是很复杂的,如果是
　　简单的,只有一两层的话,可以使用此法取到标签,然后再使用regex_extract来用正则获取下层的元素.
　　我在项目中一开始尝试的就是这个.
　　作为一个C# guy,我没有用java写过有意义的程序,最多就是hello world,所以当我决定使用java来写pig
　　UDF的时候,
　　在java项目中导入第三方jar包:
　　http://blog.iyunv.com/justinavril/article/details/2783182
　　我在浏览器中打开pig 官方文档页面和pig udf manual页面作为参考,最后再从查阅pig自带的示例脚本中得到启发.
　　整个学习的过程中,我还参照了一本书:programming pig,通过ppurl下载的.然后完成了以下的内容.
　　我还是通过xmlloader来获取最基本的结点,然后可以通过regex_extract来获取其属性.
　　然后把它作为字符串传入两个自定义的UDF,每一个udf返回的类型都是databag,然后通过flatten函数扁平化,
　　作为另外两张表的输入,传入到vertica.   
　　以上是需求的分析,其实使用C# or java 通过 dom和xpath来作还是非常简单的.
　　java工程中我写了三个类,两个类分别实现了两个UDF,都是返回databag,重写了返回的schema.
　　第三个类是一个测试驱动,里面有一个main函数来调用写的UDF,测试有没有异常.
　　java项目开发完毕后,需要编译和打包,以下代码是示例,注意编译的时候也需要把pig...jar加进去.
　　编译成类并打包:
　　[hadoop@namenode test]$ ls
　　GetDataLogs.java GetStepNodes.java
　　[hadoop@namenode test]$ javac -cp ../pig-0.12.0.jar GetStepNodes.java
　　[hadoop@namenode test]$ javac -cp ../pig-0.12.0.jar GetDataLogs.java
　　[hadoop@namenode test]$ ls
　　GetDataLogs.class GetDataLogs.java GetStepNodes.class GetStepNodes.java
　　[hadoop@namenode test]$ cd ..
　　[hadoop@namenode src]$ ls
　　pig-0.12.0.jar test
　　[hadoop@namenode src]$ jar -cf test.jar test
　　[hadoop@namenode src]$ ls
　　pig-0.12.0.jar test test.jar
　　以下是pig latin代码示例,仅供参考:





register /home/hadoop/workspace/test/src/test.jar
register /home/hadoop/pig-0.12.0/contrib/piggybank/java/piggybank.jar
xml = load '/FFA/FFA_TEST.xml' using org.apache.pig.piggybank.storage.XMLLoader('CIMProjectResults') as(testrun:chararray);
stepnodes = foreach xml generate flatten(test.GetStepNodes(testrun));
dump stepnodes;
datalognodes = foreach xml generate flatten(test.GetDataLogs(testrun));
dump datalognodes;
View Code 　　JAVA示例代码如下:





package test;
import java.io.IOException;
import java.io.StringReader;
import org.apache.pig.EvalFunc;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import javax.xml.parsers.*;
import org.w3c.dom.*;
import org.xml.sax.InputSource;
import org.xml.sax.SAXException;
import javax.xml.xpath.*;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.data.DataType;
public class GetDataLogs extends EvalFunc{
TupleFactory mTupleFactory = TupleFactory.getInstance();
BagFactory   mBagFactory   = BagFactory.getInstance();
private DataBag GetDataLogNodes(Tuple input) throws SAXException, IOException, ParserConfigurationException, XPathExpressionException
{
DataBag  output = mBagFactory.newDefaultBag();
try {
String xml = input.get(0).toString();
DocumentBuilderFactory domFactory = DocumentBuilderFactory.newInstance();
domFactory.setNamespaceAware(true);
DocumentBuilder builder = domFactory.newDocumentBuilder();
Document doc = builder.parse(new InputSource(new StringReader(xml)));
XPathFactory factory = XPathFactory.newInstance();
XPath xpath = factory.newXPath();
XPathExpression expr = xpath.compile("CIMProjectResults/Sequence");
Object result = expr.evaluate(doc, XPathConstants.NODESET);
NodeList nodes = (NodeList) result;        
NodeList stepNodes = nodes.item(0).getChildNodes();
if(stepNodes == null) return null;
for(int i=0;i