You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@stormcrawler.apache.org by jn...@apache.org on 2024/07/25 16:50:31 UTC
(incubator-stormcrawler) branch main updated: #626: Add routing field in metadata - Solr StatusUpdaterBolt (#1242)
This is an automated email from the ASF dual-hosted git repository.
jnioche pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-stormcrawler.git
The following commit(s) were added to refs/heads/main by this push:
new cb02ce95 #626: Add routing field in metadata - Solr StatusUpdaterBolt (#1242)
cb02ce95 is described below
commit cb02ce95b6b97a3e7e01ae1a58956bc655fc37b1
Author: Markos Volikas <11...@users.noreply.github.com>
AuthorDate: Thu Jul 25 19:50:26 2024 +0300
#626: Add routing field in metadata - Solr StatusUpdaterBolt (#1242)
* #626 add routing field in metadata
* #626 add dynamic field in status core
* #626 minor fixes
* #626 remove doRouting
* #626 remove unused variable
---
external/solr/cores/status/conf/schema.xml | 2 ++
external/solr/solr-conf.yaml | 6 +++-
.../org/apache/stormcrawler/solr/Constants.java | 22 ++++++++++++
.../solr/persistence/StatusUpdaterBolt.java | 40 ++++++++++++++++++++++
4 files changed, 69 insertions(+), 1 deletion(-)
diff --git a/external/solr/cores/status/conf/schema.xml b/external/solr/cores/status/conf/schema.xml
index ccdd2377..8ed1207b 100755
--- a/external/solr/cores/status/conf/schema.xml
+++ b/external/solr/cores/status/conf/schema.xml
@@ -25,5 +25,7 @@ under the License.
<field name="status" type="string" indexed="true" stored="true" required="true"/>
<field name="nextFetchDate" type="pdate" stored="true" indexed="true"/>
<dynamicField name="metadata*" type="string" indexed="true" stored="true" multiValued="true"/>
+ <field name="key" type="string" stored="true" indexed="true" multiValued="false"/>
+ <dynamicField name="*" type="string" indexed="true" stored="true" multiValued="false"/>
<uniqueKey>url</uniqueKey>
</schema>
diff --git a/external/solr/solr-conf.yaml b/external/solr/solr-conf.yaml
index 3de807be..b9d310da 100644
--- a/external/solr/solr-conf.yaml
+++ b/external/solr/solr-conf.yaml
@@ -24,7 +24,11 @@ config:
solr.status.bucket.maxsize: 5
solr.status.metadata.prefix: metadata
solr.status.max.results: 10
-
+
+ # the routing is done on the value of 'partition.url.mode'
+ # stores the value used for grouping the URLs as a separate field
+ solr.status.routing.fieldname: "key"
+
# time in secs for which the URLs will be considered for fetching after a ack or fail
spout.ttl.purgatory: 30
diff --git a/external/solr/src/main/java/org/apache/stormcrawler/solr/Constants.java b/external/solr/src/main/java/org/apache/stormcrawler/solr/Constants.java
new file mode 100644
index 00000000..57d58b54
--- /dev/null
+++ b/external/solr/src/main/java/org/apache/stormcrawler/solr/Constants.java
@@ -0,0 +1,22 @@
+/*
+ * 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.stormcrawler.solr;
+
+public interface Constants {
+
+ String PARAMPREFIX = "solr.";
+}
diff --git a/external/solr/src/main/java/org/apache/stormcrawler/solr/persistence/StatusUpdaterBolt.java b/external/solr/src/main/java/org/apache/stormcrawler/solr/persistence/StatusUpdaterBolt.java
index d05c806f..707e5571 100644
--- a/external/solr/src/main/java/org/apache/stormcrawler/solr/persistence/StatusUpdaterBolt.java
+++ b/external/solr/src/main/java/org/apache/stormcrawler/solr/persistence/StatusUpdaterBolt.java
@@ -21,6 +21,7 @@ import java.util.Iterator;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
+import org.apache.commons.lang.StringUtils;
import org.apache.solr.common.SolrInputDocument;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
@@ -28,8 +29,10 @@ import org.apache.storm.tuple.Tuple;
import org.apache.stormcrawler.Metadata;
import org.apache.stormcrawler.persistence.AbstractStatusUpdaterBolt;
import org.apache.stormcrawler.persistence.Status;
+import org.apache.stormcrawler.solr.Constants;
import org.apache.stormcrawler.solr.SolrConnection;
import org.apache.stormcrawler.util.ConfUtils;
+import org.apache.stormcrawler.util.URLPartitioner;
import org.apache.stormcrawler.util.URLUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -43,6 +46,14 @@ public class StatusUpdaterBolt extends AbstractStatusUpdaterBolt {
private static final String SolrMetadataPrefix = "solr.status.metadata.prefix";
+ private static final String SolrStatusRoutingFieldParamName =
+ Constants.PARAMPREFIX + "%s.routing.fieldname";
+
+ private URLPartitioner partitioner;
+
+ /** Store the key used for routing explicitly as a field in metadata * */
+ private String fieldNameForRoutingKey = null;
+
private String mdPrefix;
private SolrConnection connection;
@@ -55,6 +66,25 @@ public class StatusUpdaterBolt extends AbstractStatusUpdaterBolt {
mdPrefix = ConfUtils.getString(stormConf, SolrMetadataPrefix, "metadata");
+ partitioner = new URLPartitioner();
+ partitioner.configure(stormConf);
+
+ fieldNameForRoutingKey =
+ ConfUtils.getString(
+ stormConf,
+ String.format(
+ Locale.ROOT,
+ StatusUpdaterBolt.SolrStatusRoutingFieldParamName,
+ BOLT_TYPE));
+
+ if (StringUtils.isNotBlank(fieldNameForRoutingKey)) {
+ if (fieldNameForRoutingKey.startsWith("metadata.")) {
+ fieldNameForRoutingKey = fieldNameForRoutingKey.substring("metadata.".length());
+ fieldNameForRoutingKey =
+ String.format(Locale.ROOT, "%s.%s", mdPrefix, fieldNameForRoutingKey);
+ }
+ }
+
try {
connection = SolrConnection.getConnection(stormConf, BOLT_TYPE);
} catch (Exception e) {
@@ -83,6 +113,16 @@ public class StatusUpdaterBolt extends AbstractStatusUpdaterBolt {
doc.setField(String.format(Locale.ROOT, "%s.%s", mdPrefix, key), values);
}
+ String partitionKey = partitioner.getPartition(url, metadata);
+ if (partitionKey == null) {
+ partitionKey = "_DEFAULT_";
+ }
+
+ // store routing key?
+ if (StringUtils.isNotBlank(fieldNameForRoutingKey)) {
+ doc.setField(fieldNameForRoutingKey, partitionKey);
+ }
+
if (nextFetch.isPresent()) {
doc.setField("nextFetchDate", nextFetch.get());
}