Lucene reduce

[MB Leasing <le...@marketingbrokers.com>]

Hello,

For those interested, you can filter and
search Lucene documents in the reduce.

code:

import java.io.*;
import java.util.*;

import org.apache.lucene.index.Term;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.RAMDirectory;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.standard.StandardAnalyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.search.Searcher;
import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.search.BooleanQuery;
import org.apache.lucene.search.BooleanClause;
import org.apache.lucene.search.TermQuery;
import org.apache.lucene.search.Hits;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.*;

public class ql
    {
    /**************************************
    * Query Lucene using keys.
    *
    * input:
    * java^this page is about java
    * ruby^site only mentions rails
    * php^another resource about php
    * java^ejb3 discussed and spring
    * eof^eof
    *
    * make docs,search,mapreduce
    *
    * output:
    * php^topic^another resource about php
    * java^topic^this page is about java
    ***************************************/

    public static class M extends MapReduceBase implements Mapper
       {
       HashMap hm=new HashMap();
       Map group_m=Collections.synchronizedMap(hm);
       String ITEM_KEY,BATCH_KEY="";int batch=0;

       public void map(WritableComparable wc,Writable w,
          OutputCollector out,Reporter rep)throws IOException
          {
          String ln=((Text)w).toString();
          String[] parse_a=ln.split("\\^");

          if(batch>(100-1)) // new lucene document group
             {out.collect(new Text(BATCH_KEY),new BytesWritable(ob 
(group_m)));
             BATCH_KEY="BATCH_"+key_maker(String.valueOf 
(batch));batch=0;group_m.clear();}
          else if(parse_a[0].equals("eof"))out.collect(new Text 
(BATCH_KEY),new BytesWritable(ob(group_m)));
          else ;

	 ITEM_KEY="ITEM_"+key_maker(parse_a[0]);
	 Document single_d=make_lucene_doc(parse_a[0],parse_a[1],ITEM_KEY);
	 group_m.put(ITEM_KEY,single_d);
	 batch++;
          }
       }

    public static class R extends MapReduceBase implements Reducer
       {
       public void reduce(WritableComparable wc,Iterator it,
          OutputCollector out,Reporter rep)throws IOException
          {
          while(it.hasNext())
             {
	    try
	       {
	       Map m=(Map)bo(((BytesWritable)it.next()).get());
                if(m instanceof Map)
                   {
                   try
                      {
	             // build temp index
	             Directory rd=new RAMDirectory();
                      Analyzer sa=new StandardAnalyzer();
                      IndexWriter iw=new IndexWriter(rd,sa,true);

	             // unwrap,cast,send to mem
                      List keys=new ArrayList(m.keySet());
                      Iterator itr_u=keys.iterator();
                      while(itr_u.hasNext())
                         {
                         Object k_u=itr_u.next();
                         Document dtmp=(Document)m.get(k_u);
                         iw.addDocument(dtmp);
                         }

                      iw.optimize();iw.close();
                      Searcher is=new IndexSearcher(rd);

	             // simple doc filter
                      Iterator itr_s=keys.iterator();
                      while(itr_s.hasNext())
                         {
                         Object k_s=itr_s.next();
	                String tmp_topic=k_s.toString();
	                TermQuery tq_i=new TermQuery(new Term 
("item",tmp_topic.trim()));

	                // query term from key
	                tmp_topic=tmp_topic.substring(tmp_topic.lastIndexOf 
("_")+1,tmp_topic.length());
	                TermQuery tq_b=new TermQuery(new Term 
("body",tmp_topic));

	                // search topic with inventory key
                         BooleanQuery bq=new BooleanQuery();
                         bq.add(tq_i,BooleanClause.Occur.MUST);
                         bq.add(tq_b,BooleanClause.Occur.MUST);

                         Hits h=is.search(bq);
                         for(int j=0;j<h.length();j++)
                            {
                            Document doc=h.doc(j);
                            String tmp_tpc=doc.get("topic");
                            String tmp_bdy=doc.get("body");
	                   out.collect(wc,new Text(tmp_tpc+"^topic^"+tmp_bdy));
                            }
                         }
                      keys.clear();is.close();
                      }
                   catch(Exception e){System.out.println(e);}
                   }
	       }
             catch(Exception io){io.printStackTrace();}
             }
          }
       }

    public static void main(String[] args)throws Exception
       {
       String IN_DIR="/opt/hadoop-0.15.0/indir";
       String OUT_DIR="/opt/hadoop-0.15.0/outdir";

       try
          {
          JobConf jc=new JobConf(ql.class);
          jc.setJobName("ql");
          jc.setMapperClass(M.class);
          jc.setReducerClass(R.class);
          jc.setOutputKeyClass(Text.class);
          jc.setOutputValueClass(BytesWritable.class);
          jc.setInputPath(new Path(IN_DIR));
          jc.setOutputPath(new Path(OUT_DIR));
          JobClient.runJob(jc);
	 }
       catch(Exception e){System.out.println(e);}
       }

    private static Document make_lucene_doc(String in_tpc,String  
in_bdy,String in_itm)
       {
       Document d=new Document();
       d.add(new Field 
("topic",in_tpc,Field.Store.YES,Field.Index.TOKENIZED));
       d.add(new Field 
("item",in_itm,Field.Store.NO,Field.Index.UN_TOKENIZED));
       d.add(new Field 
("body",in_bdy,Field.Store.YES,Field.Index.TOKENIZED));
       return d;
       }

    private static String key_maker(String t)
       {
       Random r=new Random();
       Date d=new Date(System.currentTimeMillis());
       return key_filter(Double.toString(r.nextDouble())+d.toString() 
+"_"+t);
       }

    private static String key_filter(String s)
       { // adapted
       String rs="";StringBuffer sb=new StringBuffer("");
       String  
ok="_abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
       for(int i=0;i<s.length();i++)if(ok.indexOf(s.charAt(i))> -1) 
sb.append(s.charAt(i));
       rs=(sb.toString()).trim();sb.delete(0,sb.length());return rs;
       }

    private static Object bo(byte[] b)throws Exception
       { // adapted
       ByteArrayInputStream bis=new ByteArrayInputStream(b);
       ObjectInputStream ois=new ObjectInputStream(bis);
       Object o=ois.readObject();ois.close();bis.close();return o;
       }

    private static byte[] ob(Object o)
       { // adapted
       byte[] b=new String("").getBytes();
       ByteArrayOutputStream bos=new ByteArrayOutputStream();
       try{ObjectOutputStream os=new ObjectOutputStream(bos);
       os.writeObject(o);b=bos.toByteArray();bos.close();os.close();}
       catch(IOException e){}return b;
       }
    }

Regards,

Peter W.