如何将Lucene索引写入Hadoop1.x的HDFS系统

如何将Lucene索引写入Hadoop1.x的HDFS系统，本篇散仙将介绍上将索引写在Hadoop2.x的HDFS上，写入2.x的Hadoop相对1.x的Hadoop来说要简单的说了，因为默认solr(4.4之后的版本)里面自带的HDFSDirectory就是支持2.x的而不支持1.x的，使用2.x的Hadoop平台，可以直接把solr的corejar包拷贝到工程里面，即可使用建索引，散仙，是在eclipse上使用eclipse插件来运行hadoop程序，具体要用到的jar包，除了需要用到hadoop2.2的所有jar包外，还需增加lucene和solr的部分jar包，截图如下，散仙本次使用的是Lucene4.8.1的版本：



具体的代码如下：

Java代码  

1. package com.mapreduceindex;  
2. import org.apache.hadoop.conf.Configuration;  
3. import org.apache.hadoop.fs.FileSystem;  
4. import org.apache.hadoop.fs.Path;  
5. import org.apache.lucene.analysis.Analyzer;  
6. import org.apache.lucene.analysis.standard.StandardAnalyzer;  
7. import org.apache.lucene.document.Document;  
8. import org.apache.lucene.document.Field.Store;  
9. import org.apache.lucene.document.StringField;  
10. import org.apache.lucene.document.TextField;  
11. import org.apache.lucene.index.DirectoryReader;  
12. import org.apache.lucene.index.IndexReader;  
13. import org.apache.lucene.index.IndexWriter;  
14. import org.apache.lucene.index.IndexWriterConfig;  
15. import org.apache.lucene.index.Term;  
16. import org.apache.lucene.queryparser.classic.QueryParser;  
17. import org.apache.lucene.search.IndexSearcher;  
18. import org.apache.lucene.search.Query;  
19. import org.apache.lucene.search.ScoreDoc;  
20. import org.apache.lucene.search.TopDocs;  
21. import org.apache.lucene.store.Directory;  
22.   
23. import org.apache.lucene.util.Version;  
24. import org.apache.solr.store.hdfs.HdfsDirectory;  
25.    
26. import org.wltea.analyzer.lucene.IKAnalyzer;  
27.   
28.   
29.    
30.   
31. /** 
32.  *  
33.  * 将索引存储在Hadoop2.2的HDFS上 
34.  * 
35.  * @author qindongliang 
36.  * QQ技术交流群： 
37.  * 1号群： 324714439 如果满员了请加2号群 
38.  * 2号群： 206247899 
39.  *  
40.  *  
41.  * **/  
42. public class MyIndex {  
43.   
44.     public static  void createFile()throws Exception{  
45.           
46.           
47.         Configuration conf=new Configuration();  
48.          FileSystem fs=FileSystem.get(conf);    
49.          Path p =new Path("hdfs://192.168.46.32:9000/root/abc.txt");    
50.          fs.createNewFile(p);    
51.          //fs.create(p);    
52.          fs.close();//释放资源    
53.          System.out.println("创建文件成功.....");    
54.           
55.     }  
56.       
57.       
58.     public static void main(String[] args)throws Exception {  
59.         //createFile();  
60.         //long a=System.currentTimeMillis();  
61.        // add();  
62.         // long b=System.currentTimeMillis();  
63.         // System.out.println("耗时: "+(b-a)+"毫秒");  
64.             query("8");  
65.         //delete("3");//删除指定ID的数据  
66.     }  
67.       
68.       
69.       
70.     /*** 
71.      * 得到HDFS的writer 
72.      *  
73.      * **/  
74.     public static IndexWriter  getIndexWriter() throws Exception{  
75.           
76.         Analyzer  analyzer=new IKAnalyzer(true);  
77.         IndexWriterConfig    config=new IndexWriterConfig(Version.LUCENE_48, analyzer);  
78.         Configuration conf=new Configuration();  
79.        
80.         conf.set("fs.defaultFS","hdfs://192.168.46.32:9000/");  
81.         //conf.set("mapreduce.framework.name", "yarn");    
82.         //conf.set("yarn.resourcemanager.address", "192.168.46.32:8032");   
83.         //Path p1 =new Path("hdfs://10.2.143.5:9090/root/myfile/my.txt");  
84.         //Path path=new Path("hdfs://10.2.143.5:9090/root/myfile");  
85.         Path path=new Path("hdfs://192.168.46.32:9000/qin/myindex");  
86.         //HdfsDirectory directory=new HdfsDirectory(path, conf);  
87.           
88.         HdfsDirectory directory=new HdfsDirectory(path, conf);  
89.           
90.         IndexWriter writer=new IndexWriter(directory, config);  
91.           
92.         return writer;  
93.           
94.     }  
95.       
96.       
97.     public static void add()throws Exception{  
98.             
99.         IndexWriter writer=getIndexWriter();      
100.            
101. //      Document doc=new Document();  
102. //      doc.add(new StringField("id", "3", Store.YES));  
103. //      doc.add(new StringField("name", "lucene是一款非常优秀的全文检索框架", Store.YES));  
104. //      doc.add(new TextField("content", "我们的工资都不高", Store.YES));  
105. //      Document doc2=new Document();  
106. //      doc2.add(new StringField("id", "4", Store.YES));  
107. //      doc2.add(new StringField("name", "今天天气不错呀", Store.YES));  
108. //      doc2.add(new TextField("content", "钱存储在银行靠谱吗", Store.YES));  
109. //        
110. //      Document doc3=new Document();  
111. //      doc3.add(new StringField("id", "5", Store.YES));  
112. //      doc3.add(new StringField("name", "没有根的野草，飘忽的命途！", Store.YES));  
113. //      doc3.add(new TextField("content", "你工资多少呀！", Store.YES));  
114. //       writer.addDocument(doc);  
115. //       writer.addDocument(doc2);  
116. //      writer.addDocument(doc3);  
117.         for(int i=6;i<10000;i++){  
118.             Document doc=new Document();  
119.             doc.add(new StringField("id", i+"", Store.YES));  
120.             doc.add(new StringField("name", "lucene是一款非常优秀的全文检索框架"+i, Store.YES));  
121.             doc.add(new TextField("content", "今天发工资了吗"+i, Store.YES));  
122.             writer.addDocument(doc);  
123.             if(i%1000==0){  
124.                 writer.commit();  
125.             }  
126.         }  
127.         //writer.forceMerge(1);  
128.         writer.commit();  
129.         System.out.println("索引3条数据添加成功!");  
130.          writer.close();  
131.     }  
132.       
133.     /*** 
134.      * 添加索引 
135.      *  
136.      * **/  
137.     public static void add(Document d)throws Exception{  
138.         
139.         IndexWriter writer=getIndexWriter();      
140.         writer.addDocument(d);  
141.          writer.forceMerge(1);  
142.         writer.commit();  
143.         System.out.println("索引10000条数据添加成功!");  
144.         writer.close();  
145.     }  
146.       
147.     /** 
148.      * 根据指定ID 
149.      * 删除HDFS上的一些数据 
150.      *  
151.      *  
152.      * **/  
153.     public static void delete(String id)throws Exception{  
154.           
155.           
156.         IndexWriter writer=getIndexWriter();  
157.         writer.deleteDocuments(new Term("id", id));//删除指定ID的数据  
158.         writer.forceMerge(1);//清除已经删除的索引空间  
159.         writer.commit();//提交变化  
160.           
161.         System.out.println("id为"+id+"的数据已经删除成功.........");  
162.           
163.           
164.     }  
165.       
166.       
167.     public static void query(String queryTerm)throws Exception{  
168.         System.out.println("本次检索内容:  "+queryTerm);  
169.         Configuration conf=new Configuration();  
170.         conf.set("fs.defaultFS","hdfs://192.168.46.32:9000/");  
171.         //Path p1 =new Path("hdfs://10.2.143.5:9090/root/myfile/my.txt");  
172.     //  Path path=new Path("hdfs://192.168.75.130:9000/root/index");  
173.         Path path=new Path("hdfs://192.168.46.32:9000/qin/myindex");  
174.         Directory directory=new HdfsDirectory(path, conf);  
175.         IndexReader reader=DirectoryReader.open(directory);  
176.         System.out.println("总数据量: "+reader.numDocs());  
177.         long a=System.currentTimeMillis();  
178.         IndexSearcher searcher=new IndexSearcher(reader);  
179.         QueryParser parse=new QueryParser(Version.LUCENE_48, "content", new IKAnalyzer(true));  
180.           
181.          Query query=parse.parse(queryTerm);  
182.           
183.          TopDocs docs=searcher.search(query, 100);  
184.            
185.      System.out.println("本次命中结果:   "+docs.totalHits+"  条" );  
186.          for(ScoreDoc sc:docs.scoreDocs){  
187.              System.out.println("评分:  "+sc.score+"  id : "+searcher.doc(sc.doc).get("id")+"  name:   "+searcher.doc(sc.doc).get("name")+"   字段内容: "+searcher.doc(sc.doc).get("content"));  
188.                
189.          }  
190.         long b=System.currentTimeMillis();  
191.         System.out.println("第一次耗时:"+(b-a)+" 毫秒");  
192.     //  System.out.println("============================================");  
193. //      long c=System.currentTimeMillis();  
194. //         query=parse.parse(queryTerm);  
195. //            
196. //         docs=searcher.search(query, 100);  
197. //       System.out.println("本次命中结果:   "+docs.totalHits+"  条" );  
198. //       for(ScoreDoc sc:docs.scoreDocs){  
199. //             
200. //           System.out.println("评分:  "+sc.score+"  id : "+searcher.doc(sc.doc).get("id")+"  name:   "+searcher.doc(sc.doc).get("name")+"   字段内容: "+searcher.doc(sc.doc).get("content"));  
201. //             
202. //       }  
203. //      long d=System.currentTimeMillis();  
204. //      System.out.println("第二次耗时:"+(d-c)+" 毫秒");  
205.           
206.          reader.close();  
207.          directory.close();  
208.            
209.          System.out.println("检索完毕...............");  
210.        
211.           
212.           
213.           
214.     }  
215.       
216.       
217.       
218.       
219. }  

package com.mapreduceindex;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
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.Store;
import org.apache.lucene.document.StringField;
import org.apache.lucene.document.TextField;
import org.apache.lucene.index.DirectoryReader;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.IndexWriterConfig;
import org.apache.lucene.index.Term;
import org.apache.lucene.queryparser.classic.QueryParser;
import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.search.Query;
import org.apache.lucene.search.ScoreDoc;
import org.apache.lucene.search.TopDocs;
import org.apache.lucene.store.Directory;

import org.apache.lucene.util.Version;
import org.apache.solr.store.hdfs.HdfsDirectory;
 
import org.wltea.analyzer.lucene.IKAnalyzer;


 

/**
 * 
 * 将索引存储在Hadoop2.2的HDFS上
 *
 * @author qindongliang
 * QQ技术交流群：
 * 1号群： 324714439 如果满员了请加2号群
 * 2号群： 206247899
 * 
 * 
 * **/
public class MyIndex {

	public static  void createFile()throws Exception{
		
		
		Configuration conf=new Configuration();
		 FileSystem fs=FileSystem.get(conf);  
         Path p =new Path("hdfs://192.168.46.32:9000/root/abc.txt");  
         fs.createNewFile(p);  
         //fs.create(p);  
         fs.close();//释放资源  
         System.out.println("创建文件成功.....");  
		
	}
	
	
	public static void main(String[] args)throws Exception {
		//createFile();
		//long a=System.currentTimeMillis();
	   // add();
	 	// long b=System.currentTimeMillis();
	 	// System.out.println("耗时: "+(b-a)+"毫秒");
		    query("8");
		//delete("3");//删除指定ID的数据
	}
	
	
	
	/***
	 * 得到HDFS的writer
	 * 
	 * **/
	public static IndexWriter  getIndexWriter() throws Exception{
		
		Analyzer  analyzer=new IKAnalyzer(true);
 		IndexWriterConfig    config=new IndexWriterConfig(Version.LUCENE_48, analyzer);
 		Configuration conf=new Configuration();
 	 
 		conf.set("fs.defaultFS","hdfs://192.168.46.32:9000/");
		//conf.set("mapreduce.framework.name", "yarn");  
		//conf.set("yarn.resourcemanager.address", "192.168.46.32:8032"); 
 		//Path p1 =new Path("hdfs://10.2.143.5:9090/root/myfile/my.txt");
 		//Path path=new Path("hdfs://10.2.143.5:9090/root/myfile");
 		Path path=new Path("hdfs://192.168.46.32:9000/qin/myindex");
 		//HdfsDirectory directory=new HdfsDirectory(path, conf);
 		
 		HdfsDirectory directory=new HdfsDirectory(path, conf);
 		
 		IndexWriter writer=new IndexWriter(directory, config);
 		
 		return writer;
		
	}
	
	
	public static void add()throws Exception{
	      
		IndexWriter writer=getIndexWriter();	
		 
//		Document doc=new Document();
//		doc.add(new StringField("id", "3", Store.YES));
//		doc.add(new StringField("name", "lucene是一款非常优秀的全文检索框架", Store.YES));
//		doc.add(new TextField("content", "我们的工资都不高", Store.YES));
//		Document doc2=new Document();
//		doc2.add(new StringField("id", "4", Store.YES));
//		doc2.add(new StringField("name", "今天天气不错呀", Store.YES));
//		doc2.add(new TextField("content", "钱存储在银行靠谱吗", Store.YES));
//		
//		Document doc3=new Document();
//		doc3.add(new StringField("id", "5", Store.YES));
//		doc3.add(new StringField("name", "没有根的野草，飘忽的命途！", Store.YES));
//		doc3.add(new TextField("content", "你工资多少呀！", Store.YES));
// 		 writer.addDocument(doc);
//		 writer.addDocument(doc2);
//		writer.addDocument(doc3);
		for(int i=6;i<10000;i++){
			Document doc=new Document();
			doc.add(new StringField("id", i+"", Store.YES));
			doc.add(new StringField("name", "lucene是一款非常优秀的全文检索框架"+i, Store.YES));
			doc.add(new TextField("content", "今天发工资了吗"+i, Store.YES));
			writer.addDocument(doc);
			if(i%1000==0){
				writer.commit();
			}
		}
		//writer.forceMerge(1);
		writer.commit();
		System.out.println("索引3条数据添加成功!");
		 writer.close();
	}
	
	/***
	 * 添加索引
	 * 
	 * **/
	public static void add(Document d)throws Exception{
      
		IndexWriter writer=getIndexWriter();	
		writer.addDocument(d);
		 writer.forceMerge(1);
		writer.commit();
		System.out.println("索引10000条数据添加成功!");
		writer.close();
	}
	
	/**
	 * 根据指定ID
	 * 删除HDFS上的一些数据
	 * 
	 * 
	 * **/
	public static void delete(String id)throws Exception{
		
		
		IndexWriter writer=getIndexWriter();
		writer.deleteDocuments(new Term("id", id));//删除指定ID的数据
		writer.forceMerge(1);//清除已经删除的索引空间
		writer.commit();//提交变化
		
		System.out.println("id为"+id+"的数据已经删除成功.........");
		
		
	}
	
	
	public static void query(String queryTerm)throws Exception{
		System.out.println("本次检索内容:  "+queryTerm);
		Configuration conf=new Configuration();
		conf.set("fs.defaultFS","hdfs://192.168.46.32:9000/");
 		//Path p1 =new Path("hdfs://10.2.143.5:9090/root/myfile/my.txt");
 	//	Path path=new Path("hdfs://192.168.75.130:9000/root/index");
 		Path path=new Path("hdfs://192.168.46.32:9000/qin/myindex");
		Directory directory=new HdfsDirectory(path, conf);
		IndexReader reader=DirectoryReader.open(directory);
		System.out.println("总数据量: "+reader.numDocs());
		long a=System.currentTimeMillis();
		IndexSearcher searcher=new IndexSearcher(reader);
		QueryParser parse=new QueryParser(Version.LUCENE_48, "content", new IKAnalyzer(true));
		
		 Query query=parse.parse(queryTerm);
		
		 TopDocs docs=searcher.search(query, 100);
		 
 	 System.out.println("本次命中结果:   "+docs.totalHits+"  条" );
		 for(ScoreDoc sc:docs.scoreDocs){
			 System.out.println("评分:  "+sc.score+"  id : "+searcher.doc(sc.doc).get("id")+"  name:   "+searcher.doc(sc.doc).get("name")+"   字段内容: "+searcher.doc(sc.doc).get("content"));
			 
		 }
		long b=System.currentTimeMillis();
		System.out.println("第一次耗时:"+(b-a)+" 毫秒");
	//	System.out.println("============================================");
//		long c=System.currentTimeMillis();
//		   query=parse.parse(queryTerm);
//			
//		   docs=searcher.search(query, 100);
//		 System.out.println("本次命中结果:   "+docs.totalHits+"  条" );
//		 for(ScoreDoc sc:docs.scoreDocs){
//			 
//			 System.out.println("评分:  "+sc.score+"  id : "+searcher.doc(sc.doc).get("id")+"  name:   "+searcher.doc(sc.doc).get("name")+"   字段内容: "+searcher.doc(sc.doc).get("content"));
//			 
//		 }
//		long d=System.currentTimeMillis();
//		System.out.println("第二次耗时:"+(d-c)+" 毫秒");
		
		 reader.close();
		 directory.close();
		 
		 System.out.println("检索完毕...............");
	 
		
		
		
	}
	
	
	
	
}

使用IK的分词器，建立索引完毕后，在HDFS上的索引如下截图：


检索数据时，第一次检索往往比较慢，第一次之后因为有了Block Cache，所以第二次，检索的速度非常快，当然这也跟你机器的配置有关系：

Java代码  

1. 本次检索内容:  8  
2. WARN - NativeCodeLoader.<clinit>(62) | Unable to load native-hadoop library for your platform... using builtin-java classes where applicable  
3. 总数据量: 9994  
4. 本次命中结果:   1  条  
5. 评分:  4.7582965  id : 8  name:   lucene是一款非常优秀的全文检索框架8   字段内容: 今天发工资了吗8  
6. 第一次耗时:261 毫秒  
7. ============================================  
8. 本次命中结果:   1  条  
9. 评分:  4.7582965  id : 8  name:   lucene是一款非常优秀的全文检索框架8   字段内容: 今天发工资了吗8  
10. 第二次耗时:6 毫秒  
11. INFO - HdfsDirectory.close(97) | Closing hdfs directory hdfs://192.168.46.32:9000/qin/myindex  
12. 检索完毕...............  

本次检索内容:  8
WARN - NativeCodeLoader.<clinit>(62) | Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
总数据量: 9994
本次命中结果:   1  条
评分:  4.7582965  id : 8  name:   lucene是一款非常优秀的全文检索框架8   字段内容: 今天发工资了吗8
第一次耗时:261 毫秒
============================================
本次命中结果:   1  条
评分:  4.7582965  id : 8  name:   lucene是一款非常优秀的全文检索框架8   字段内容: 今天发工资了吗8
第二次耗时:6 毫秒
INFO - HdfsDirectory.close(97) | Closing hdfs directory hdfs://192.168.46.32:9000/qin/myindex
检索完毕...............

为什么要使用Hadoop建索引？ 使用Hadoop建索引可以利用MapReduce分布式计算能力从而大大提升建索引的速度，这一点优势很明显，但美中不足的是在Hadoop上做检索，性能却不怎么好，虽然有了块缓存，但是如果索引被按64M的块被切分到不同的节点上，那么检索的时候，就需要跨机器从各个块上扫描，拉取命中数据，这一点是很耗时的，目前，据散仙所知，还没有比较好的部署在Hadoop上的分布式检索方案，但毫无疑问的是建索引的能力，确实很给力，后面散仙会写如何使用MapReduce来并行构建Lucene索引，其实既然单机版的都可以完成，那么稍微改造下变成MapReduce作业