如何使用Apache Pig与Lucene集成，还不知道的道友们，可以先看下上篇，熟悉下具体的流程。

package com.pig.support.solr;import java.io.IOException;import java.util.ArrayList;import java.util.List;import java.util.concurrent.Executors;import java.util.concurrent.ScheduledExecutorService;import java.util.concurrent.TimeUnit;import org.apache.hadoop.io.Writable;import org.apache.hadoop.mapreduce.JobContext;import org.apache.hadoop.mapreduce.OutputCommitter;import org.apache.hadoop.mapreduce.RecordWriter;import org.apache.hadoop.mapreduce.TaskAttemptContext;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.solr.client.solrj.SolrServer;import org.apache.solr.client.solrj.SolrServerException;import org.apache.solr.client.solrj.impl.CloudSolrServer;import org.apache.solr.client.solrj.impl.HttpSolrServer;import org.apache.solr.common.SolrInputDocument;/** * @author qindongliang * 支持SOlr的SolrOutputFormat * 如果你想了解，或学习更多这方面的 * 知识，请加入我们的群： *  * 搜索技术交流群（2000人）：324714439  * 大数据技术1号交流群（2000人）：376932160  （已满） * 大数据技术2号交流群（2000人）：415886155  * 微信公众号：我是攻城师（woshigcs） *  * */public class SolrOutputFormat extends		FileOutputFormat
    
      {	final String address;	final String collection;	public SolrOutputFormat(String address, String collection) {		this.address = address;		this.collection = collection;	}	@Override	public RecordWriter
     
       getRecordWriter(			TaskAttemptContext ctx) throws IOException, InterruptedException {		return new SolrRecordWriter(ctx, address, collection);	}		@Override	public synchronized OutputCommitter getOutputCommitter(			TaskAttemptContext arg0) throws IOException {		return new OutputCommitter(){			@Override			public void abortTask(TaskAttemptContext ctx) throws IOException {							}			@Override			public void commitTask(TaskAttemptContext ctx) throws IOException {							}			@Override			public boolean needsTaskCommit(TaskAttemptContext arg0)					throws IOException {				return true;			}			@Override			public void setupJob(JobContext ctx) throws IOException {							}			@Override			public void setupTask(TaskAttemptContext ctx) throws IOException {							}								};	}	/**	 * Write out the LuceneIndex to a local temporary location.
      
	 * On commit/close the index is copied to the hdfs output directory.
      
	 * 	 */	static class SolrRecordWriter extends RecordWriter
      
        {		/**Solr的地址*/		SolrServer server;		/**批处理提交的数量**/		int batch = 5000;				TaskAttemptContext ctx;				List
       
         docs = new ArrayList
        
         (batch);		ScheduledExecutorService exec = Executors.newSingleThreadScheduledExecutor();		/**		 * Opens and forces connect to CloudSolrServer		 * 		 * @param address		 */		public SolrRecordWriter(final TaskAttemptContext ctx, String address, String collection) {			try {				this.ctx = ctx;				server = new HttpSolrServer(address);								exec.scheduleWithFixedDelay(new Runnable(){					public void run(){						ctx.progress();					}				}, 1000, 1000, TimeUnit.MILLISECONDS);			} catch (Exception e) {				RuntimeException exc = new RuntimeException(e.toString(), e);				exc.setStackTrace(e.getStackTrace());				throw exc;			}		}				/**		 * On close we commit		 */		@Override		public void close(final TaskAttemptContext ctx) throws IOException,				InterruptedException {			try {								if (docs.size() > 0) {					server.add(docs);					docs.clear();				}				server.commit();			} catch (SolrServerException e) {				RuntimeException exc = new RuntimeException(e.toString(), e);				exc.setStackTrace(e.getStackTrace());				throw exc;			} finally {				server.shutdown();				exec.shutdownNow();			}					}		/**		 * We add the indexed documents without commit		 */		@Override		public void write(Writable key, SolrInputDocument doc)				throws IOException, InterruptedException {			try {				docs.add(doc);				if (docs.size() >= batch) {					server.add(docs);					docs.clear();				}			} catch (SolrServerException e) {				RuntimeException exc = new RuntimeException(e.toString(), e);				exc.setStackTrace(e.getStackTrace());				throw exc;			}		}	}}
        
       
      
     
    

package com.pig.support.solr;import java.io.IOException;import java.util.Properties;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.Writable;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.OutputFormat;import org.apache.hadoop.mapreduce.RecordWriter;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.pig.ResourceSchema;import org.apache.pig.ResourceSchema.ResourceFieldSchema;import org.apache.pig.ResourceStatistics;import org.apache.pig.StoreFunc;import org.apache.pig.StoreMetadata;import org.apache.pig.data.Tuple;import org.apache.pig.impl.util.UDFContext;import org.apache.pig.impl.util.Utils;import org.apache.solr.common.SolrInputDocument;/** *  * Create a lucene index *  */public class SolrStore extends StoreFunc implements StoreMetadata {	private static final String SCHEMA_SIGNATURE = "solr.output.schema";	ResourceSchema schema;	String udfSignature;	RecordWriter
    
      writer;	String address;	String collection;		public SolrStore(String address, String collection) {		this.address = address;		this.collection = collection;	}	public void storeStatistics(ResourceStatistics stats, String location,			Job job) throws IOException {	}	public void storeSchema(ResourceSchema schema, String location, Job job)			throws IOException {	}	@Override	public void checkSchema(ResourceSchema s) throws IOException {		UDFContext udfc = UDFContext.getUDFContext();		Properties p = udfc.getUDFProperties(this.getClass(),				new String[] { udfSignature });		p.setProperty(SCHEMA_SIGNATURE, s.toString());	}	public OutputFormat
     
       getOutputFormat()			throws IOException {		// not be used		return new SolrOutputFormat(address, collection);	}	/**	 * Not used	 */	@Override	public void setStoreLocation(String location, Job job) throws IOException {		FileOutputFormat.setOutputPath(job, new Path(location));	}	@Override	public void setStoreFuncUDFContextSignature(String signature) {		this.udfSignature = signature;	}	@SuppressWarnings({ "unchecked", "rawtypes" })	@Override	public void prepareToWrite(RecordWriter writer) throws IOException {		this.writer = writer;		UDFContext udc = UDFContext.getUDFContext();		String schemaStr = udc.getUDFProperties(this.getClass(),				new String[] { udfSignature }).getProperty(SCHEMA_SIGNATURE);		if (schemaStr == null) {			throw new RuntimeException("Could not find udf signature");		}		schema = new ResourceSchema(Utils.getSchemaFromString(schemaStr));	}	/**	 * Shamelessly copied from : https://issues.apache.org/jira/secure/attachment/12484764/NUTCH-1016-2.0.patch	 * @param input	 * @return	 */	private static String stripNonCharCodepoints(String input) {		StringBuilder retval = new StringBuilder(input.length());		char ch;		for (int i = 0; i < input.length(); i++) {			ch = input.charAt(i);			// Strip all non-characters			// http://unicode.org/cldr/utility/list-unicodeset.jsp?a=[:Noncharacter_Code_Point=True:]			// and non-printable control characters except tabulator, new line			// and carriage return			if (ch % 0x10000 != 0xffff && // 0xffff - 0x10ffff range step											// 0x10000					ch % 0x10000 != 0xfffe && // 0xfffe - 0x10fffe range					(ch <= 0xfdd0 || ch >= 0xfdef) && // 0xfdd0 - 0xfdef					(ch > 0x1F || ch == 0x9 || ch == 0xa || ch == 0xd)) {				retval.append(ch);			}		}		return retval.toString();	}	@Override	public void putNext(Tuple t) throws IOException {		final SolrInputDocument doc = new SolrInputDocument();		final ResourceFieldSchema[] fields = schema.getFields();		int docfields = 0;		for (int i = 0; i < fields.length; i++) {			final Object value = t.get(i);			if (value != null) {				docfields++;				doc.addField(fields[i].getName().trim(), stripNonCharCodepoints(value.toString()));			}		}		try {			if (docfields > 0)				writer.write(null, doc);		} catch (InterruptedException e) {			Thread.currentThread().interrupt();			return;		}	}}
     
    

--注册依赖文件的jar包REGISTER ./dependfiles/tools.jar;--注册solr相关的jar包REGISTER  ./solrdependfiles/pigudf.jar; REGISTER  ./solrdependfiles/solr-core-4.10.2.jar;REGISTER  ./solrdependfiles/solr-solrj-4.10.2.jar;REGISTER  ./solrdependfiles/httpclient-4.3.1.jarREGISTER  ./solrdependfiles/httpcore-4.3.jarREGISTER  ./solrdependfiles/httpmime-4.3.1.jarREGISTER  ./solrdependfiles/noggit-0.5.jar--加载HDFS数据，并定义scheamla = load '/tmp/data' using PigStorage(',') as (sword:chararray,scount:int);--存储到solr中，并提供solr的ip地址和端口号store d into '/user/search/solrindextemp'  using com.pig.support.solr.SolrStore('http://localhost:8983/solr/collection1','collection1');~                                                                                                                                                            ~                                                                      ~