diff --git build.xml build.xml index f6c5b8e..cd6f00f 100644 --- build.xml +++ build.xml @@ -934,6 +934,7 @@ --> + diff --git conf/nutch-default.xml conf/nutch-default.xml index 9269cd8..6c712ef 100644 --- conf/nutch-default.xml +++ conf/nutch-default.xml @@ -1178,6 +1178,27 @@ + + solr.zookeeper.hosts + + A list of one or more Zookeeper ensembles to write + updates. Each ensemble is + configured as host:port|host:port and + ensembles are separated by a comma. When + indexing to + two separate + SolrCloud clusters e.g. + zoo_a_1:2181|zoo_a_2:2181,zoo_b_1:2181|zoo_b_2:2181 + + + + + solr.collection + + The SolrCloud collection to write updates to. + + + elastic.host diff --git src/plugin/build.xml src/plugin/build.xml index 3c8df80..528178c 100755 --- src/plugin/build.xml +++ src/plugin/build.xml @@ -32,6 +32,7 @@ + @@ -119,6 +120,7 @@ + diff --git src/plugin/indexer-solrcloud/build.xml src/plugin/indexer-solrcloud/build.xml new file mode 100644 index 0000000..be0aa87 --- /dev/null +++ src/plugin/indexer-solrcloud/build.xml @@ -0,0 +1,22 @@ + + + + + + + diff --git src/plugin/indexer-solrcloud/ivy.xml src/plugin/indexer-solrcloud/ivy.xml new file mode 100644 index 0000000..c80d436 --- /dev/null +++ src/plugin/indexer-solrcloud/ivy.xml @@ -0,0 +1,43 @@ + + + + + + + + + + Apache Nutch + + + + + + + + + + + + + + + + + diff --git src/plugin/indexer-solrcloud/plugin.xml src/plugin/indexer-solrcloud/plugin.xml new file mode 100644 index 0000000..aba5e4c --- /dev/null +++ src/plugin/indexer-solrcloud/plugin.xml @@ -0,0 +1,54 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git src/plugin/indexer-solrcloud/src/java/org/apache/nutch/indexwriter/solrcloud/SolrCloudConstants.java src/plugin/indexer-solrcloud/src/java/org/apache/nutch/indexwriter/solrcloud/SolrCloudConstants.java new file mode 100644 index 0000000..6acdd09 --- /dev/null +++ src/plugin/indexer-solrcloud/src/java/org/apache/nutch/indexwriter/solrcloud/SolrCloudConstants.java @@ -0,0 +1,39 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nutch.indexwriter.solrcloud; + +public interface SolrCloudConstants { + public static final String SOLR_PREFIX = "solr."; + + public static final String ZOOKEEPER_HOSTS = SOLR_PREFIX + "zookeeper.hosts"; + + public static final String COMMIT_SIZE = SOLR_PREFIX + "commit.size"; + + public static final String COLLECTION = SOLR_PREFIX + "collection"; + + public static final String MAPPING_FILE = SOLR_PREFIX + "mapping.file"; + + public static final String USE_AUTH = SOLR_PREFIX + "auth"; + + public static final String USERNAME = SOLR_PREFIX + "auth.username"; + + public static final String PASSWORD = SOLR_PREFIX + "auth.password"; + + @Deprecated + public static final String PARAMS = SOLR_PREFIX + "params"; + +} diff --git src/plugin/indexer-solrcloud/src/java/org/apache/nutch/indexwriter/solrcloud/SolrCloudIndexWriter.java src/plugin/indexer-solrcloud/src/java/org/apache/nutch/indexwriter/solrcloud/SolrCloudIndexWriter.java new file mode 100644 index 0000000..332445b --- /dev/null +++ src/plugin/indexer-solrcloud/src/java/org/apache/nutch/indexwriter/solrcloud/SolrCloudIndexWriter.java @@ -0,0 +1,180 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nutch.indexwriter.solrcloud; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map.Entry; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.apache.hadoop.conf.Configuration; +import org.apache.nutch.indexer.NutchDocument; +import org.apache.nutch.indexer.IndexWriter; +import org.apache.solr.client.solrj.SolrServerException; +import org.apache.solr.client.solrj.impl.CloudSolrServer; +import org.apache.solr.common.SolrInputDocument; + +public class SolrCloudIndexWriter implements IndexWriter { + + public static final Logger LOG = LoggerFactory + .getLogger(SolrCloudIndexWriter.class); + + private CloudSolrServer solr; + private SolrCloudMappingReader solrMapping; + + private Configuration config; + + private final List inputDocs = new ArrayList(); + + private int batchSize; + private int numDeletes = 0; + private boolean delete = false; + + protected static long documentCount = 0; + + @Override + public void open(Configuration conf) throws IOException { + solr = SolrCloudUtils.getSolrServer(conf); + batchSize = conf.getInt(SolrCloudConstants.COMMIT_SIZE, 1000); + solrMapping = SolrCloudMappingReader.getInstance(conf); + } + + @Override + public void write(NutchDocument doc) throws IOException { + final SolrInputDocument inputDoc = new SolrInputDocument(); + for (final Entry> e : doc) { + for (final String val : e.getValue()) { + + Object val2 = val; + if (e.getKey().equals("content") || e.getKey().equals("title")) { + val2 = SolrCloudUtils.stripNonCharCodepoints(val); + } + + inputDoc.addField(solrMapping.mapKey(e.getKey()), val2); + String sCopy = solrMapping.mapCopyKey(e.getKey()); + if (sCopy != e.getKey()) { + inputDoc.addField(sCopy, val2); + } + } + } + inputDoc.setDocumentBoost(doc.getScore()); + inputDocs.add(inputDoc); + documentCount++; + if (inputDocs.size() >= batchSize) { + try { + LOG.info("Adding " + Integer.toString(inputDocs.size()) + " documents"); + solr.add(inputDocs); + } catch (final SolrServerException e) { + throw new IOException(e); + } + inputDocs.clear(); + } + } + + @Override + public void close() throws IOException { + try { + if (!inputDocs.isEmpty()) { + LOG.info("Adding " + Integer.toString(inputDocs.size()) + " documents"); + solr.add(inputDocs); + inputDocs.clear(); + } else if (numDeletes > 0) { + LOG.info("Deleted " + Integer.toString(numDeletes) + " documents"); + } + } catch (final SolrServerException e) { + throw new IOException(e); + } + } + + @Override + public Configuration getConf() { + return config; + } + + @Override + public void setConf(Configuration conf) { + config = conf; + String serverURL = conf.get(SolrCloudConstants.ZOOKEEPER_HOSTS); + if (serverURL == null) { + String message = "Missing SOLR Zookeeper URL. Should be set via -D " + + SolrCloudConstants.ZOOKEEPER_HOSTS; + message += "\n" + describe(); + LOG.error(message); + throw new RuntimeException(message); + } + } + + @Override + public void delete(String key) throws IOException { + if (delete) { + try { + solr.deleteById(key); + numDeletes++; + } catch (final SolrServerException e) { + throw makeIOException(e); + } + } + } + + @Override + public void update(NutchDocument doc) throws IOException { + write(doc); + } + + @Override + public void commit() throws IOException { + try { + solr.commit(); + LOG.info("Total " + documentCount + + (documentCount > 1 ? " documents are " : " document is ") + + "added."); + } catch (SolrServerException e) { + throw makeIOException(e); + } + } + + public static IOException makeIOException(SolrServerException e) { + final IOException ioe = new IOException(); + ioe.initCause(e); + return ioe; + } + + @Override + public String describe() { + StringBuffer sb = new StringBuffer("SOLRCloudIndexWriter\n"); + sb.append("\t").append(SolrCloudConstants.ZOOKEEPER_HOSTS) + .append(" : URL of the SOLR Zookeeper Server (mandatory)\n"); + sb.append("\t").append(SolrCloudConstants.COMMIT_SIZE) + .append(" : buffer size when sending to SOLR (default 1000)\n"); + sb.append("\t") + .append(SolrCloudConstants.MAPPING_FILE) + .append( + " : name of the mapping file for fields (default solrindex-mapping.xml)\n"); + sb.append("\t").append(SolrCloudConstants.USE_AUTH) + .append(" : use authentication (default false)\n"); + sb.append("\t").append(SolrCloudConstants.USERNAME) + .append(" : use authentication (default false)\n"); + sb.append("\t").append(SolrCloudConstants.USE_AUTH) + .append(" : username for authentication\n"); + sb.append("\t").append(SolrCloudConstants.PASSWORD) + .append(" : password for authentication\n"); + return sb.toString(); + } + +} diff --git src/plugin/indexer-solrcloud/src/java/org/apache/nutch/indexwriter/solrcloud/SolrCloudMappingReader.java src/plugin/indexer-solrcloud/src/java/org/apache/nutch/indexwriter/solrcloud/SolrCloudMappingReader.java new file mode 100644 index 0000000..e89a005 --- /dev/null +++ src/plugin/indexer-solrcloud/src/java/org/apache/nutch/indexwriter/solrcloud/SolrCloudMappingReader.java @@ -0,0 +1,149 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nutch.indexwriter.solrcloud; + +import java.io.IOException; +import java.io.InputStream; +import java.net.MalformedURLException; +import java.util.HashMap; +import java.util.Map; + +import javax.xml.parsers.DocumentBuilder; +import javax.xml.parsers.DocumentBuilderFactory; +import javax.xml.parsers.ParserConfigurationException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.apache.hadoop.conf.Configuration; +import org.apache.nutch.util.ObjectCache; +import org.w3c.dom.Document; +import org.w3c.dom.Element; +import org.w3c.dom.NodeList; +import org.xml.sax.InputSource; +import org.xml.sax.SAXException; + +public class SolrCloudMappingReader { + public static Logger LOG = LoggerFactory + .getLogger(SolrCloudMappingReader.class); + + private Configuration conf; + + private Map keyMap = new HashMap(); + private Map copyMap = new HashMap(); + private String uniqueKey = "id"; + + public static synchronized SolrCloudMappingReader getInstance( + Configuration conf) { + ObjectCache cache = ObjectCache.get(conf); + SolrCloudMappingReader instance = (SolrCloudMappingReader) cache + .getObject(SolrCloudMappingReader.class.getName()); + if (instance == null) { + instance = new SolrCloudMappingReader(conf); + cache.setObject(SolrCloudMappingReader.class.getName(), instance); + } + return instance; + } + + protected SolrCloudMappingReader(Configuration conf) { + this.conf = conf; + parseMapping(); + } + + private void parseMapping() { + InputStream ssInputStream = null; + ssInputStream = conf.getConfResourceAsInputStream(conf.get( + SolrCloudConstants.MAPPING_FILE, "solrindex-mapping.xml")); + + InputSource inputSource = new InputSource(ssInputStream); + try { + DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance(); + DocumentBuilder builder = factory.newDocumentBuilder(); + Document document = builder.parse(inputSource); + Element rootElement = document.getDocumentElement(); + NodeList fieldList = rootElement.getElementsByTagName("field"); + if (fieldList.getLength() > 0) { + for (int i = 0; i < fieldList.getLength(); i++) { + Element element = (Element) fieldList.item(i); + LOG.info("source: " + element.getAttribute("source") + " dest: " + + element.getAttribute("dest")); + keyMap.put(element.getAttribute("source"), + element.getAttribute("dest")); + } + } + NodeList copyFieldList = rootElement.getElementsByTagName("copyField"); + if (copyFieldList.getLength() > 0) { + for (int i = 0; i < copyFieldList.getLength(); i++) { + Element element = (Element) copyFieldList.item(i); + LOG.info("source: " + element.getAttribute("source") + " dest: " + + element.getAttribute("dest")); + copyMap.put(element.getAttribute("source"), + element.getAttribute("dest")); + } + } + NodeList uniqueKeyItem = rootElement.getElementsByTagName("uniqueKey"); + if (uniqueKeyItem.getLength() > 1) { + LOG.warn("More than one unique key definitions found in solr index mapping, using default 'id'"); + uniqueKey = "id"; + } else if (uniqueKeyItem.getLength() == 0) { + LOG.warn("No unique key definition found in solr index mapping using, default 'id'"); + } else { + uniqueKey = uniqueKeyItem.item(0).getFirstChild().getNodeValue(); + } + } catch (MalformedURLException e) { + LOG.warn(e.toString()); + } catch (SAXException e) { + LOG.warn(e.toString()); + } catch (IOException e) { + LOG.warn(e.toString()); + } catch (ParserConfigurationException e) { + LOG.warn(e.toString()); + } + } + + public Map getKeyMap() { + return keyMap; + } + + public Map getCopyMap() { + return copyMap; + } + + public String getUniqueKey() { + return uniqueKey; + } + + public String hasCopy(String key) { + if (copyMap.containsKey(key)) { + key = copyMap.get(key); + } + return key; + } + + public String mapKey(String key) throws IOException { + if (keyMap.containsKey(key)) { + key = keyMap.get(key); + } + return key; + } + + public String mapCopyKey(String key) throws IOException { + if (copyMap.containsKey(key)) { + key = copyMap.get(key); + } + return key; + } +} diff --git src/plugin/indexer-solrcloud/src/java/org/apache/nutch/indexwriter/solrcloud/SolrCloudUtils.java src/plugin/indexer-solrcloud/src/java/org/apache/nutch/indexwriter/solrcloud/SolrCloudUtils.java new file mode 100644 index 0000000..8288cfb --- /dev/null +++ src/plugin/indexer-solrcloud/src/java/org/apache/nutch/indexwriter/solrcloud/SolrCloudUtils.java @@ -0,0 +1,116 @@ +package org.apache.nutch.indexwriter.solrcloud; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.apache.hadoop.conf.Configuration; +import org.apache.solr.client.solrj.SolrServer; +import org.apache.solr.client.solrj.impl.CloudSolrServer; + +import java.net.MalformedURLException; +import java.util.ArrayList; +import java.util.List; + +/** + * Handy Solr utility used for creating SolrJ clients and cleaning the input. + */ +public class SolrCloudUtils { + + public static Logger LOG = LoggerFactory.getLogger(SolrCloudUtils.class); + + public static CloudSolrServer getSolrServer(Configuration job) + throws MalformedURLException { + List ss = getSolrServers(job); + return ss.get(0); + } + + /** + * Returns a list of SolrServers based on configuration. It will either create + * a list of single node HttpSolrServers or a list of CloudSolrServers. Write, + * deletes and updates are sent to all SolrServers. + * + * @param Configuration + * @return SolrServer + */ + public static List getSolrServers(Configuration job) + throws MalformedURLException { + String[] zookeeper = job.getStrings(SolrCloudConstants.ZOOKEEPER_HOSTS); + + ArrayList ss = new ArrayList(); + + // Got URL's for a Zookeeper ensemble? We'll prefer that then! + if (zookeeper != null && zookeeper.length > 0) { + for (int i = 0; i < zookeeper.length; i++) { + try { + CloudSolrServer server = getCloudServer( + zookeeper[i].replace('|', ','), job); + ss.add(server); + } catch (Exception e) { + LOG.warn(e.getMessage()); + } + } + } else { + LOG.warn("SolrCloudIndexer: Zookeeper Hosts do not set."); + } + return ss; + } + + /** + * Shuts down the specified CloudSolrServer implementation. + * + * @param SolrServer + */ + public static void shutdown(SolrServer solrServer) { + if (solrServer instanceof CloudSolrServer) { + solrServer.shutdown(); + } + } + + /** + * Get a CloudSolrServer for the given Zookeeper URL. + * + * @param String + * @param Configuration + * @return SolrServer + */ + private static CloudSolrServer getCloudServer(String url, Configuration job) + throws MalformedURLException { + LOG.info("SolrCloudIndexer: opening Zookeeper connection to: " + url); + + CloudSolrServer ss = new CloudSolrServer(url); + ss.setDefaultCollection(job.get(SolrCloudConstants.COLLECTION)); + ss.connect(); + return ss; + } + + /** + * Strips unicode non-char codepoints from the input string. Text content sent + * to Solr sometimes contains crap characters. Stripping them from the input + * makes it possible to transmit them without issues. + * + * @param String + * input + * @return Sting + */ + public static String stripNonCharCodepoints(String input) { + StringBuilder retval = new StringBuilder(); + 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(); + } +}