mirror of
https://github.com/internetarchive/heritrix3.git
synced 2026-09-26 15:46:31 +00:00
Merge pull request #621 from nla/remove-hbase
Remove HBase from contrib
This commit is contained in:
Alex Osborne
@@ -24,41 +24,6 @@
|
||||
<version>${project.version}</version>
|
||||
<scope>compile</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.hbase</groupId>
|
||||
<artifactId>hbase-client</artifactId>
|
||||
<version>0.98.6-cdh5.3.5</version>
|
||||
<exclusions>
|
||||
<exclusion>
|
||||
<artifactId>jets3t</artifactId>
|
||||
<groupId>net.java.dev.jets3t</groupId>
|
||||
</exclusion>
|
||||
<exclusion>
|
||||
<artifactId>junit</artifactId>
|
||||
<groupId>junit</groupId>
|
||||
</exclusion>
|
||||
<!-- tools.jar is not available in JDK 11 so exclude it
|
||||
hbase-client accidentally leaked it as a transitive dependency
|
||||
https://issues.apache.org/jira/browse/HBASE-13963 -->
|
||||
<exclusion>
|
||||
<groupId>jdk.tools</groupId>
|
||||
<artifactId>jdk.tools</artifactId>
|
||||
</exclusion>
|
||||
<!-- prefer guava 17 from webarchive-commons over guava 12 from hbase-client -->
|
||||
<exclusion>
|
||||
<groupId>com.google.guava</groupId>
|
||||
<artifactId>guava</artifactId>
|
||||
</exclusion>
|
||||
<exclusion>
|
||||
<groupId>commons-io</groupId>
|
||||
<artifactId>commons-io</artifactId>
|
||||
</exclusion>
|
||||
<exclusion>
|
||||
<groupId>commons-collections</groupId>
|
||||
<artifactId>commons-collections</artifactId>
|
||||
</exclusion>
|
||||
</exclusions>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.rabbitmq</groupId>
|
||||
<artifactId>amqp-client</artifactId>
|
||||
|
||||
@@ -18,10 +18,6 @@
|
||||
*/
|
||||
package org.archive.modules.recrawl;
|
||||
|
||||
import java.text.DateFormat;
|
||||
import java.text.ParseException;
|
||||
import java.text.SimpleDateFormat;
|
||||
import java.util.Date;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
@@ -82,35 +78,7 @@ public class FetchHistoryHelper {
|
||||
return null;
|
||||
}
|
||||
|
||||
protected static final DateFormat HTTP_DATE_FORMAT = new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss zzz");
|
||||
|
||||
protected FetchHistoryHelper() {
|
||||
}
|
||||
|
||||
/**
|
||||
* converts time in HTTP Date format {@code dateStr} to seconds
|
||||
* since epoch.
|
||||
* @param dateStr time in HTTP Date format.
|
||||
* @return seconds since epoch
|
||||
*/
|
||||
public static long parseHttpDate(String dateStr) {
|
||||
synchronized (HTTP_DATE_FORMAT) {
|
||||
try {
|
||||
Date d = HTTP_DATE_FORMAT.parse(dateStr);
|
||||
return d.getTime() / 1000;
|
||||
} catch (ParseException ex) {
|
||||
if (logger.isDebugEnabled())
|
||||
logger.debug("bad HTTP DATE: " + dateStr);
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public static String formatHttpDate(long time) {
|
||||
synchronized (HTTP_DATE_FORMAT) {
|
||||
// format(Date) is not thread safe either
|
||||
return HTTP_DATE_FORMAT.format(new Date(time * 1000));
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,126 +0,0 @@
|
||||
/*
|
||||
* This file is part of the Heritrix web crawler (crawler.archive.org).
|
||||
*
|
||||
* Licensed to the Internet Archive (IA) by one or more individual
|
||||
* contributors.
|
||||
*
|
||||
* The IA 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.archive.modules.recrawl.hbase;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Map;
|
||||
import java.util.Map.Entry;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.hadoop.hbase.HBaseConfiguration;
|
||||
import org.apache.hadoop.hbase.client.HBaseAdmin;
|
||||
import org.springframework.context.Lifecycle;
|
||||
|
||||
/**
|
||||
* Represents a deployment of HBase. (An instance, a database, an HBase...)
|
||||
*
|
||||
* @author nlevitt
|
||||
*/
|
||||
public class HBase implements Lifecycle {
|
||||
|
||||
private static final Logger logger =
|
||||
Logger.getLogger(HBase.class.getName());
|
||||
|
||||
protected Configuration conf = null;
|
||||
|
||||
private Map<String,String> properties;
|
||||
|
||||
public Map<String,String> getProperties() {
|
||||
return properties;
|
||||
}
|
||||
|
||||
public void setProperties(Map<String,String> properties) {
|
||||
this.properties = properties;
|
||||
|
||||
if (conf == null) {
|
||||
conf = HBaseConfiguration.create();
|
||||
}
|
||||
for (Entry<String, String> entry: getProperties().entrySet()) {
|
||||
conf.set(entry.getKey(), entry.getValue());
|
||||
}
|
||||
}
|
||||
|
||||
public synchronized Configuration configuration() {
|
||||
if (conf == null) {
|
||||
conf = HBaseConfiguration.create();
|
||||
}
|
||||
|
||||
return conf;
|
||||
}
|
||||
|
||||
// convenience setters
|
||||
public void setZookeeperQuorum(String value) {
|
||||
configuration().set("hbase.zookeeper.quorum", value);
|
||||
}
|
||||
public void setZookeeperClientPort(int port) {
|
||||
configuration().setInt("hbase.zookeeper.property.clientPort", port);
|
||||
}
|
||||
|
||||
protected transient HBaseAdmin admin;
|
||||
|
||||
public synchronized HBaseAdmin admin() throws IOException {
|
||||
if (admin == null) {
|
||||
admin = new HBaseAdmin(configuration());
|
||||
}
|
||||
|
||||
return admin;
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void stop() {
|
||||
isRunning = false;
|
||||
if (admin != null) {
|
||||
try {
|
||||
admin.close();
|
||||
} catch (IOException e) {
|
||||
logger.warning("problem closing HBaseAdmin " + admin + " - " + e);
|
||||
}
|
||||
|
||||
admin = null;
|
||||
}
|
||||
if (conf != null) {
|
||||
// HConnectionManager.deleteConnection(conf); // XXX?
|
||||
conf = null;
|
||||
}
|
||||
}
|
||||
|
||||
protected transient boolean isRunning = false;
|
||||
@Override
|
||||
public boolean isRunning() {
|
||||
return isRunning;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void start() {
|
||||
isRunning = true;
|
||||
}
|
||||
|
||||
public synchronized void reset() {
|
||||
if (admin != null) {
|
||||
try {
|
||||
admin.close();
|
||||
} catch (IOException e) {
|
||||
logger.warning("problem closing HBaseAdmin " + admin + " - " + e);
|
||||
}
|
||||
|
||||
admin = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
-303
@@ -1,303 +0,0 @@
|
||||
package org.archive.modules.recrawl.hbase;
|
||||
|
||||
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_CONTENT_DIGEST_COUNT;
|
||||
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_ORIGINAL_DATE;
|
||||
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_ORIGINAL_URL;
|
||||
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_WARC_FILENAME;
|
||||
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_WARC_FILE_OFFSET;
|
||||
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_WARC_RECORD_ID;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.HashMap;
|
||||
import java.util.Iterator;
|
||||
import java.util.Map;
|
||||
import java.util.Map.Entry;
|
||||
import java.util.logging.Level;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
import org.apache.hadoop.hbase.HColumnDescriptor;
|
||||
import org.apache.hadoop.hbase.HTableDescriptor;
|
||||
import org.apache.hadoop.hbase.client.Get;
|
||||
import org.apache.hadoop.hbase.client.HBaseAdmin;
|
||||
import org.apache.hadoop.hbase.client.Put;
|
||||
import org.apache.hadoop.hbase.client.Result;
|
||||
import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
|
||||
import org.apache.hadoop.hbase.regionserver.NoSuchColumnFamilyException;
|
||||
import org.apache.hadoop.hbase.util.Bytes;
|
||||
import org.archive.modules.CrawlURI;
|
||||
import org.archive.modules.recrawl.AbstractContentDigestHistory;
|
||||
import org.json.JSONException;
|
||||
import org.json.JSONObject;
|
||||
import org.springframework.context.Lifecycle;
|
||||
|
||||
import com.google.common.collect.BiMap;
|
||||
import com.google.common.collect.HashBiMap;
|
||||
|
||||
/**
|
||||
* HBase content digest history store. Must be a toplevel bean in
|
||||
* crawler-beans.cxml in order to receive {@link Lifecycle} events.
|
||||
*
|
||||
* @see AbstractContentDigestHistory
|
||||
* @author nlevitt
|
||||
*/
|
||||
public class HBaseContentDigestHistory extends AbstractContentDigestHistory implements Lifecycle {
|
||||
|
||||
private static final Logger logger =
|
||||
Logger.getLogger(HBaseContentDigestHistory.class.getName());
|
||||
|
||||
protected static final byte[] COLUMN_FAMILY = Bytes.toBytes("f");
|
||||
protected static final byte[] COLUMN = Bytes.toBytes("c");
|
||||
|
||||
protected static final BiMap<String,String> JSON_KEYS_MAP = HashBiMap.create();
|
||||
static {
|
||||
JSON_KEYS_MAP.put(A_CONTENT_DIGEST_COUNT, "c");
|
||||
JSON_KEYS_MAP.put(A_ORIGINAL_URL, "u");
|
||||
JSON_KEYS_MAP.put(A_WARC_RECORD_ID, "i");
|
||||
JSON_KEYS_MAP.put(A_WARC_FILENAME, "f");
|
||||
JSON_KEYS_MAP.put(A_WARC_FILE_OFFSET, "o");
|
||||
JSON_KEYS_MAP.put(A_ORIGINAL_DATE, "d");
|
||||
}
|
||||
|
||||
protected HBaseTable table;
|
||||
public void setTable(HBaseTable table) {
|
||||
this.table = table;
|
||||
}
|
||||
|
||||
protected boolean addColumnFamily = false;
|
||||
public boolean getAddColumnFamily() {
|
||||
return addColumnFamily;
|
||||
}
|
||||
/**
|
||||
* Add the expected column family
|
||||
* {@link #COLUMN_FAMILY} to the HBase table if the
|
||||
* table doesn't already have it.
|
||||
*/
|
||||
public void setAddColumnFamily(boolean addColumnFamily) {
|
||||
this.addColumnFamily = addColumnFamily;
|
||||
}
|
||||
|
||||
protected int retryIntervalMs = 10*1000;
|
||||
public int getRetryIntervalMs() {
|
||||
return retryIntervalMs;
|
||||
}
|
||||
public void setRetryIntervalMs(int retryIntervalMs) {
|
||||
this.retryIntervalMs = retryIntervalMs;
|
||||
}
|
||||
|
||||
protected int maxTries = 1;
|
||||
public int getMaxTries() {
|
||||
return maxTries;
|
||||
}
|
||||
public void setMaxTries(int maxTries) {
|
||||
this.maxTries = maxTries;
|
||||
}
|
||||
|
||||
protected String keySuffix = null;
|
||||
public String getKeySuffix() {
|
||||
return keySuffix;
|
||||
}
|
||||
|
||||
/**
|
||||
* If not null, keySuffix is appended to the lookup key when loading and
|
||||
* storing digest history. Thus the key looks like {digest}{keySuffix}, e.g.
|
||||
* "sha1:22SFHXERHNFOEY6WK7YOUN4PFIPZSB4D-1193". The purpose is to support
|
||||
* multiple namespaces in a single hbase table, to avoid proliferation of
|
||||
* small tables. The reason we use a suffix instead of a prefix is to leave
|
||||
* open the possibility of deduplication across these different namespaces
|
||||
* at some point in the future.
|
||||
*
|
||||
* @param keySuffix
|
||||
*/
|
||||
public void setKeySuffix(String keySuffix) {
|
||||
this.keySuffix = keySuffix;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected String persistKeyFor(CrawlURI curi) {
|
||||
if (keySuffix != null) {
|
||||
return super.persistKeyFor(curi) + keySuffix;
|
||||
} else {
|
||||
return super.persistKeyFor(curi);
|
||||
}
|
||||
}
|
||||
|
||||
protected synchronized void addColumnFamily() {
|
||||
try {
|
||||
HTableDescriptor oldDesc = table.getHtableDescriptor();
|
||||
if (oldDesc.getFamily(COLUMN_FAMILY) == null) {
|
||||
HTableDescriptor newDesc = new HTableDescriptor(oldDesc);
|
||||
newDesc.addFamily(new HColumnDescriptor(COLUMN_FAMILY));
|
||||
logger.info("table does not yet have expected column family, modifying descriptor to " + newDesc);
|
||||
HBaseAdmin hbaseAdmin = table.getHbase().admin();
|
||||
hbaseAdmin.disableTable(table.getName());
|
||||
hbaseAdmin.modifyTable(Bytes.toBytes(table.getName()), newDesc);
|
||||
hbaseAdmin.enableTable(table.getName());
|
||||
}
|
||||
} catch (IOException e) {
|
||||
logger.warning("problem adding column family: " + e);
|
||||
}
|
||||
}
|
||||
|
||||
private boolean isRunning;
|
||||
@Override
|
||||
public void start() {
|
||||
// add column family here to avoid disabling table while another
|
||||
// ToeThread is trying to use it
|
||||
if (getAddColumnFamily()) {
|
||||
addColumnFamily();
|
||||
}
|
||||
this.isRunning = true;
|
||||
}
|
||||
@Override
|
||||
public void stop() {
|
||||
this.isRunning = false;
|
||||
}
|
||||
@Override
|
||||
public boolean isRunning() {
|
||||
return isRunning;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void load(CrawlURI curi) {
|
||||
// make this call in all cases so that the value is initialized and
|
||||
// WARCWriterProcessor knows it should put the info in there
|
||||
HashMap<String, Object> contentDigestHistory = curi.getContentDigestHistory();
|
||||
|
||||
byte[] key = Bytes.toBytes(persistKeyFor(curi));
|
||||
Result hbaseResult = tryHbaseGet(curi, new Get(key));
|
||||
|
||||
if (hbaseResult != null) {
|
||||
Map<String, Object> loadedHistory = parseHbaseResult(curi, hbaseResult);
|
||||
|
||||
if (loadedHistory != null) {
|
||||
if (logger.isLoggable(Level.FINER)) {
|
||||
logger.finer("loaded history by digest " + persistKeyFor(curi)
|
||||
+ " for uri " + curi + " - " + loadedHistory);
|
||||
}
|
||||
contentDigestHistory.putAll(loadedHistory);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
protected Result tryHbaseGet(CrawlURI curi, Get hbaseGet) {
|
||||
try {
|
||||
return table.get(hbaseGet);
|
||||
} catch (IOException e) {
|
||||
logger.warning("problem retrieving persist data from hbase, proceeding without, for digest " + persistKeyFor(curi) + " uri " + curi + " - " + e);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
protected Map<String, Object> parseHbaseResult(CrawlURI curi, Result hbaseResult) {
|
||||
HashMap<String, Object> loadedHistory = null;
|
||||
// no data for uri is indicated by empty Result
|
||||
if (!hbaseResult.isEmpty()) {
|
||||
byte[] jsonBytes = hbaseResult.getValue(COLUMN_FAMILY, COLUMN);
|
||||
if (jsonBytes != null) {
|
||||
JSONObject json = null;
|
||||
try {
|
||||
json = new JSONObject(Bytes.toString(jsonBytes));
|
||||
loadedHistory = new HashMap<String,Object>();
|
||||
@SuppressWarnings("unchecked")
|
||||
Iterator<String> keyIter = json.keys();
|
||||
while (keyIter.hasNext()) {
|
||||
String jsonKey = keyIter.next();
|
||||
Object jsonValue = json.get(jsonKey);
|
||||
String historyMapKey = JSON_KEYS_MAP.inverse().get(jsonKey);
|
||||
if (historyMapKey == null) {
|
||||
logger.warning("unknown key \"" + jsonKey + "\" found in hbase json for digest " + persistKeyFor(curi));
|
||||
historyMapKey = jsonKey;
|
||||
}
|
||||
loadedHistory.put(historyMapKey, jsonValue);
|
||||
}
|
||||
} catch (JSONException e) {
|
||||
logger.warning("problem parsing json for digest " + persistKeyFor(curi) + " uri " + curi + " - " + e);
|
||||
}
|
||||
} else {
|
||||
// shouldn't happen? result.isEmpty() is normal case
|
||||
logger.fine("[jsonBytes==null] no persist data for digest " + persistKeyFor(curi) + " uri " + curi);
|
||||
}
|
||||
} else {
|
||||
logger.finest("[result.isEmpty()] no persist data for digest " + persistKeyFor(curi) + " uri " + curi);
|
||||
}
|
||||
|
||||
return loadedHistory;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void store(CrawlURI curi) {
|
||||
if (!curi.hasContentDigestHistory()
|
||||
|| curi.getContentDigestHistory().isEmpty()) {
|
||||
return;
|
||||
}
|
||||
if (logger.isLoggable(Level.FINER)) {
|
||||
logger.finer("storing history by digest " + persistKeyFor(curi)
|
||||
+ " for uri " + curi + " - "
|
||||
+ curi.getContentDigestHistory());
|
||||
}
|
||||
|
||||
Put hbasePut = createHbasePut(curi);
|
||||
tryHbasePut(curi, hbasePut);
|
||||
}
|
||||
|
||||
protected void tryHbasePut(CrawlURI curi, Put p) {
|
||||
int tryCount = 0;
|
||||
do {
|
||||
tryCount++;
|
||||
try {
|
||||
table.put(p);
|
||||
return;
|
||||
} catch (RetriesExhaustedWithDetailsException e) {
|
||||
if (e.getCause(0) instanceof NoSuchColumnFamilyException && getAddColumnFamily()) {
|
||||
addColumnFamily();
|
||||
tryCount--;
|
||||
} else {
|
||||
logger.warning("put failed " + "(try " + tryCount + " of "
|
||||
+ getMaxTries() + ")" + " for " + curi + " - " + e);
|
||||
}
|
||||
} catch (IOException e) {
|
||||
logger.warning("put failed " + "(try " + tryCount + " of "
|
||||
+ getMaxTries() + ")" + " for " + curi + " - " + e);
|
||||
} catch (NullPointerException e) {
|
||||
// HTable.put() throws NullPointerException while connection is lost.
|
||||
logger.warning("put failed " + "(try " + tryCount + " of "
|
||||
+ getMaxTries() + ")" + " for " + curi + " - " + e);
|
||||
}
|
||||
|
||||
if (tryCount > 0 && tryCount < getMaxTries() && isRunning()) {
|
||||
try {
|
||||
Thread.sleep(getRetryIntervalMs());
|
||||
} catch (InterruptedException ex) {
|
||||
logger.warning("thread interrupted. aborting retry for " + curi);
|
||||
return;
|
||||
}
|
||||
}
|
||||
} while (tryCount < getMaxTries() && isRunning());
|
||||
|
||||
if (isRunning()) {
|
||||
logger.warning("giving up after " + tryCount + " tries on put for " + curi);
|
||||
}
|
||||
}
|
||||
|
||||
protected Put createHbasePut(CrawlURI curi) {
|
||||
byte[] key = Bytes.toBytes(persistKeyFor(curi));
|
||||
Put hbasePut = new Put(key);
|
||||
try {
|
||||
JSONObject json = new JSONObject();
|
||||
for (Entry<String, Object> entry: curi.getContentDigestHistory().entrySet()) {
|
||||
String jsonKey = JSON_KEYS_MAP.get(entry.getKey());
|
||||
if (jsonKey == null) {
|
||||
logger.warning("unknown key \"" + entry.getKey() + "\" found in content digest history map for " + curi);
|
||||
jsonKey = entry.getKey();
|
||||
}
|
||||
json.put(jsonKey, entry.getValue());
|
||||
}
|
||||
hbasePut.add(COLUMN_FAMILY, COLUMN, Bytes.toBytes(json.toString()));
|
||||
} catch (JSONException e) {
|
||||
// should not happen - all values are either primitive or String.
|
||||
logger.log(Level.SEVERE, "problem creating json object for digest " + persistKeyFor(curi) + " uri " + curi, e);
|
||||
}
|
||||
return hbasePut;
|
||||
}
|
||||
|
||||
}
|
||||
-88
@@ -1,88 +0,0 @@
|
||||
/*
|
||||
* This file is part of the Heritrix web crawler (crawler.archive.org).
|
||||
*
|
||||
* Licensed to the Internet Archive (IA) by one or more individual
|
||||
* contributors.
|
||||
*
|
||||
* The IA 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.archive.modules.recrawl.hbase;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.logging.Level;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
import org.apache.hadoop.hbase.client.Get;
|
||||
import org.apache.hadoop.hbase.client.Result;
|
||||
import org.archive.modules.CrawlURI;
|
||||
import org.archive.modules.ProcessResult;
|
||||
import org.archive.modules.Processor;
|
||||
import org.archive.modules.recrawl.FetchHistoryProcessor;
|
||||
|
||||
/**
|
||||
* A {@link Processor} for retrieving recrawl info from HBase table.
|
||||
* See {@link HBasePersistProcessor} for table schema.
|
||||
* As with other fetch history processors, this needs to be combined with {@link FetchHistoryProcessor}
|
||||
* (set up after FetchHTTP, before WarcWriter) to work.
|
||||
* @see HBasePersistStoreProcessor
|
||||
* @author kenji
|
||||
*/
|
||||
public class HBasePersistLoadProcessor extends HBasePersistProcessor {
|
||||
private static final Logger logger =
|
||||
Logger.getLogger(HBasePersistLoadProcessor.class.getName());
|
||||
|
||||
@Override
|
||||
protected ProcessResult innerProcessResult(CrawlURI uri) throws InterruptedException {
|
||||
byte[] key = rowKeyForURI(uri);
|
||||
Get g = new Get(key);
|
||||
try {
|
||||
Result r = table.get(g);
|
||||
// no data for uri is indicated by empty Result
|
||||
if (r.isEmpty()) {
|
||||
if (logger.isLoggable(Level.FINE)) {
|
||||
logger.fine(uri + ": <no crawlinfo>");
|
||||
}
|
||||
return ProcessResult.PROCEED;
|
||||
}
|
||||
schema.load(r, uri);
|
||||
if (uri.getFetchStatus() < 0) {
|
||||
return ProcessResult.FINISH;
|
||||
}
|
||||
} catch (IOException e) {
|
||||
logger.warning("problem retrieving persist data from hbase, proceeding without, for " + uri + " - " + e);
|
||||
} catch (Exception ex) {
|
||||
// get() throws RuntimeException upon ZooKeeper connection failures.
|
||||
// no crawl history load failure should make fetch of URL fail.
|
||||
logger.log(Level.WARNING, "Get failed for " + uri + ": ", ex);
|
||||
}
|
||||
return ProcessResult.PROCEED;
|
||||
}
|
||||
|
||||
/**
|
||||
* unused.
|
||||
*/
|
||||
@Override
|
||||
protected void innerProcess(CrawlURI uri) throws InterruptedException {
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean shouldProcess(CrawlURI uri) {
|
||||
// TODO: we want deduplicate robots.txt, too.
|
||||
//if (uri.isPrerequisite()) return false;
|
||||
String scheme = uri.getUURI().getScheme();
|
||||
if (!(scheme.equals("http") || scheme.equals("https") || scheme.equals("ftp") || scheme.equals("sftp"))) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
}
|
||||
@@ -1,32 +0,0 @@
|
||||
package org.archive.modules.recrawl.hbase;
|
||||
|
||||
import org.archive.modules.CrawlURI;
|
||||
import org.archive.modules.recrawl.AbstractPersistProcessor;
|
||||
import org.springframework.beans.factory.annotation.Required;
|
||||
|
||||
/**
|
||||
* A base class for processors for keeping de-duplication data in HBase.
|
||||
* Table schema is defined by {@link RecrawlDataSchema} implementation.
|
||||
* @author kenji
|
||||
*/
|
||||
public abstract class HBasePersistProcessor extends AbstractPersistProcessor {
|
||||
|
||||
protected HBaseTableBean table;
|
||||
@Required
|
||||
public void setTable(HBaseTableBean table) {
|
||||
this.table = table;
|
||||
}
|
||||
|
||||
protected RecrawlDataSchema schema;
|
||||
public RecrawlDataSchema getSchema() {
|
||||
return schema;
|
||||
}
|
||||
@Required
|
||||
public void setSchema(RecrawlDataSchema schema) {
|
||||
this.schema = schema;
|
||||
}
|
||||
|
||||
protected byte[] rowKeyForURI(CrawlURI curi) {
|
||||
return schema.rowKeyForURI(curi);
|
||||
}
|
||||
}
|
||||
-133
@@ -1,133 +0,0 @@
|
||||
/*
|
||||
* This file is part of the Heritrix web crawler (crawler.archive.org).
|
||||
*
|
||||
* Licensed to the Internet Archive (IA) by one or more individual
|
||||
* contributors.
|
||||
*
|
||||
* The IA 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.archive.modules.recrawl.hbase;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
import org.apache.hadoop.hbase.HColumnDescriptor;
|
||||
import org.apache.hadoop.hbase.HTableDescriptor;
|
||||
import org.apache.hadoop.hbase.client.HBaseAdmin;
|
||||
import org.apache.hadoop.hbase.client.Put;
|
||||
import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
|
||||
import org.apache.hadoop.hbase.regionserver.NoSuchColumnFamilyException;
|
||||
import org.apache.hadoop.hbase.util.Bytes;
|
||||
import org.archive.modules.CrawlURI;
|
||||
import org.archive.modules.fetcher.FetchStatusCodes;
|
||||
import org.archive.modules.recrawl.RecrawlAttributeConstants;
|
||||
|
||||
/**
|
||||
* @author kenji
|
||||
*/
|
||||
public class HBasePersistStoreProcessor extends HBasePersistProcessor implements FetchStatusCodes, RecrawlAttributeConstants {
|
||||
private static final Logger logger = Logger.getLogger(HBasePersistStoreProcessor.class.getName());
|
||||
|
||||
protected boolean addColumnFamily = false;
|
||||
public boolean getAddColumnFamily() {
|
||||
return addColumnFamily;
|
||||
}
|
||||
/**
|
||||
* Add the expected column family
|
||||
* {@link HBaseContentDigestHistory#COLUMN_FAMILY} to the HBase table if the
|
||||
* table doesn't already have it.
|
||||
*/
|
||||
public void setAddColumnFamily(boolean addColumnFamily) {
|
||||
this.addColumnFamily = addColumnFamily;
|
||||
}
|
||||
|
||||
protected int retryIntervalMs = 10*1000;
|
||||
public int getRetryIntervalMs() {
|
||||
return retryIntervalMs;
|
||||
}
|
||||
public void setRetryIntervalMs(int retryIntervalMs) {
|
||||
this.retryIntervalMs = retryIntervalMs;
|
||||
}
|
||||
|
||||
protected int maxTries = 1;
|
||||
public int getMaxTries() {
|
||||
return maxTries;
|
||||
}
|
||||
public void setMaxTries(int maxTries) {
|
||||
this.maxTries = maxTries;
|
||||
}
|
||||
|
||||
protected synchronized void addColumnFamily() {
|
||||
try {
|
||||
HTableDescriptor oldDesc = table.getHtableDescriptor();
|
||||
byte[] columnFamily = Bytes.toBytes(schema.getColumnFamily());
|
||||
if (oldDesc.getFamily(columnFamily) == null) {
|
||||
HTableDescriptor newDesc = new HTableDescriptor(oldDesc);
|
||||
newDesc.addFamily(new HColumnDescriptor(columnFamily));
|
||||
logger.info("table does not yet have expected column family, modifying descriptor to " + newDesc);
|
||||
HBaseAdmin hbaseAdmin = table.getHbase().admin();
|
||||
hbaseAdmin.disableTable(table.getName());
|
||||
hbaseAdmin.modifyTable(Bytes.toBytes(table.getName()), newDesc);
|
||||
hbaseAdmin.enableTable(table.getName());
|
||||
}
|
||||
} catch (IOException e) {
|
||||
logger.warning("problem adding column family: " + e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void innerProcess(CrawlURI uri) {
|
||||
Put p = schema.createPut(uri);
|
||||
int tryCount = 0;
|
||||
do {
|
||||
tryCount++;
|
||||
try {
|
||||
table.put(p);
|
||||
return;
|
||||
} catch (RetriesExhaustedWithDetailsException e) {
|
||||
if (e.getCause(0) instanceof NoSuchColumnFamilyException && getAddColumnFamily()) {
|
||||
addColumnFamily();
|
||||
tryCount--;
|
||||
} else {
|
||||
logger.warning("put failed " + "(try " + tryCount + " of "
|
||||
+ getMaxTries() + ")" + " for " + uri + " - " + e);
|
||||
}
|
||||
} catch (IOException e) {
|
||||
logger.warning("put failed " + "(try " + tryCount + " of "
|
||||
+ getMaxTries() + ")" + " for " + uri + " - " + e);
|
||||
} catch (NullPointerException e) {
|
||||
// HTable.put() throws NullPointerException while connection is lost.
|
||||
logger.warning("put failed " + "(try " + tryCount + " of "
|
||||
+ getMaxTries() + ")" + " for " + uri + " - " + e);
|
||||
}
|
||||
|
||||
if (tryCount > 0 && tryCount < getMaxTries() && isRunning()) {
|
||||
try {
|
||||
Thread.sleep(getRetryIntervalMs());
|
||||
} catch (InterruptedException ex) {
|
||||
logger.warning("thread interrupted. aborting retry for " + uri);
|
||||
return;
|
||||
}
|
||||
}
|
||||
} while (tryCount < getMaxTries() && isRunning());
|
||||
|
||||
if (isRunning()) {
|
||||
logger.warning("giving up after " + tryCount + " tries on put for " + uri);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected boolean shouldProcess(CrawlURI curi) {
|
||||
return super.shouldStore(curi);
|
||||
}
|
||||
}
|
||||
@@ -1,159 +0,0 @@
|
||||
/*
|
||||
* This file is part of the Heritrix web crawler (crawler.archive.org).
|
||||
*
|
||||
* Licensed to the Internet Archive (IA) by one or more individual
|
||||
* contributors.
|
||||
*
|
||||
* The IA 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.archive.modules.recrawl.hbase;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.logging.Level;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
import org.apache.hadoop.hbase.HTableDescriptor;
|
||||
import org.apache.hadoop.hbase.TableName;
|
||||
import org.apache.hadoop.hbase.client.Get;
|
||||
import org.apache.hadoop.hbase.client.HBaseAdmin;
|
||||
import org.apache.hadoop.hbase.client.HConnection;
|
||||
import org.apache.hadoop.hbase.client.HConnectionManager;
|
||||
import org.apache.hadoop.hbase.client.HTableInterface;
|
||||
import org.apache.hadoop.hbase.client.Put;
|
||||
import org.apache.hadoop.hbase.client.Result;
|
||||
|
||||
/**
|
||||
* @author kenji
|
||||
* @author nlevitt
|
||||
*/
|
||||
public class HBaseTable extends HBaseTableBean {
|
||||
|
||||
static final Logger logger =
|
||||
Logger.getLogger(HBaseTable.class.getName());
|
||||
|
||||
protected boolean create = false;
|
||||
protected HConnection hconn = null;
|
||||
protected ThreadLocal<HTableInterface> htable = new ThreadLocal<HTableInterface>();
|
||||
|
||||
public boolean getCreate() {
|
||||
return create;
|
||||
}
|
||||
/** Create the named table if it doesn't exist. */
|
||||
public void setCreate(boolean create) {
|
||||
this.create = create;
|
||||
}
|
||||
|
||||
public HBaseTable() {
|
||||
}
|
||||
|
||||
protected synchronized HConnection hconnection() throws IOException {
|
||||
if (hconn == null) {
|
||||
hconn = HConnectionManager.createConnection(hbase.configuration());
|
||||
}
|
||||
return hconn;
|
||||
}
|
||||
|
||||
protected HTableInterface htable() throws IOException {
|
||||
if (htable.get() == null) {
|
||||
htable.set(hconnection().getTable(htableName));
|
||||
}
|
||||
return htable.get();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void put(Put p) throws IOException {
|
||||
try {
|
||||
htable().put(p);
|
||||
} catch (IOException e) {
|
||||
reset();
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Result get(Get g) throws IOException {
|
||||
try {
|
||||
return htable().get(g);
|
||||
} catch (IOException e) {
|
||||
reset();
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
public HTableDescriptor getHtableDescriptor() throws IOException {
|
||||
try {
|
||||
return htable().getTableDescriptor();
|
||||
} catch (IOException e) {
|
||||
reset();
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void start() {
|
||||
if (getCreate()) {
|
||||
int attempt = 1;
|
||||
while (true) {
|
||||
try {
|
||||
HBaseAdmin admin = hbase.admin();
|
||||
if (!admin.tableExists(htableName)) {
|
||||
HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(htableName));
|
||||
logger.info("hbase table '" + htableName + "' does not exist, creating it... " + desc);
|
||||
admin.createTable(desc);
|
||||
}
|
||||
break;
|
||||
} catch (IOException e) {
|
||||
logger.log(Level.WARNING, "(attempt " + attempt + ") problem creating hbase table " + htableName, e);
|
||||
attempt++;
|
||||
reset();
|
||||
// back off up to 60 seconds between retries
|
||||
try {
|
||||
Thread.sleep(Math.min(attempt * 1000, 60000));
|
||||
} catch (InterruptedException e1) {
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
super.start();
|
||||
}
|
||||
|
||||
protected void reset() {
|
||||
if (htable.get() != null) {
|
||||
try {
|
||||
htable.get().close();
|
||||
} catch (IOException e) {
|
||||
logger.log(Level.WARNING, "htablename='" + htableName + "' htable.close() threw " + e, e);
|
||||
}
|
||||
htable.remove();
|
||||
}
|
||||
|
||||
if (hconn != null) {
|
||||
try {
|
||||
hconn.close();
|
||||
} catch (IOException e) {
|
||||
logger.log(Level.WARNING, "hconn.close() threw " + e, e);
|
||||
}
|
||||
// HConnectionManager.deleteStaleConnection(hconn);
|
||||
hconn = null;
|
||||
}
|
||||
|
||||
hbase.reset();
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void stop() {
|
||||
super.stop();
|
||||
reset();
|
||||
}
|
||||
}
|
||||
@@ -1,78 +0,0 @@
|
||||
package org.archive.modules.recrawl.hbase;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
import org.apache.hadoop.hbase.HTableDescriptor;
|
||||
import org.apache.hadoop.hbase.client.Get;
|
||||
import org.apache.hadoop.hbase.client.Put;
|
||||
import org.apache.hadoop.hbase.client.Result;
|
||||
import org.archive.modules.recrawl.PersistOnlineProcessor;
|
||||
import org.springframework.context.Lifecycle;
|
||||
|
||||
/**
|
||||
* base class for different types of HBaseTable Spring bean implementations.
|
||||
* @author kenji
|
||||
* @author nlevitt
|
||||
*
|
||||
*/
|
||||
public abstract class HBaseTableBean implements Lifecycle {
|
||||
|
||||
protected String htableName = PersistOnlineProcessor.URI_HISTORY_DBNAME;
|
||||
protected HBase hbase = new HBase();
|
||||
protected transient boolean isRunning = false;
|
||||
|
||||
// <backward-compatibility>
|
||||
public void setName(String name) {
|
||||
this.htableName = name;
|
||||
}
|
||||
|
||||
public String getName() {
|
||||
return htableName;
|
||||
}
|
||||
// </backward-compatibility>
|
||||
|
||||
/**
|
||||
* set name of single HTable this instance accesses.
|
||||
* @param htableName
|
||||
*/
|
||||
public void setHtableName(String htableName) {
|
||||
this.htableName = htableName;
|
||||
}
|
||||
public String getHtableName() {
|
||||
return htableName;
|
||||
}
|
||||
|
||||
public HBaseTableBean() {
|
||||
super();
|
||||
}
|
||||
|
||||
public void setHbase(HBase hbase) {
|
||||
this.hbase = hbase;
|
||||
}
|
||||
|
||||
public HBase getHbase() {
|
||||
return hbase;
|
||||
}
|
||||
|
||||
public abstract void put(Put p) throws IOException;
|
||||
|
||||
public abstract Result get(Get g) throws IOException;
|
||||
|
||||
public abstract HTableDescriptor getHtableDescriptor() throws IOException;
|
||||
|
||||
@Override
|
||||
public boolean isRunning() {
|
||||
return isRunning;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void start() {
|
||||
isRunning = true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void stop() {
|
||||
isRunning = false;
|
||||
}
|
||||
|
||||
}
|
||||
-140
@@ -1,140 +0,0 @@
|
||||
/*
|
||||
* This file is part of the Heritrix web crawler (crawler.archive.org).
|
||||
*
|
||||
* Licensed to the Internet Archive (IA) by one or more individual
|
||||
* contributors.
|
||||
*
|
||||
* The IA 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.archive.modules.recrawl.hbase;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
import org.apache.hadoop.hbase.KeyValue;
|
||||
import org.apache.hadoop.hbase.client.Put;
|
||||
import org.apache.hadoop.hbase.client.Result;
|
||||
import org.apache.hadoop.hbase.util.Bytes;
|
||||
import org.archive.modules.CrawlURI;
|
||||
import org.archive.modules.fetcher.FetchStatusCodes;
|
||||
import org.archive.modules.recrawl.FetchHistoryHelper;
|
||||
import org.archive.modules.recrawl.RecrawlAttributeConstants;
|
||||
|
||||
/**
|
||||
* RecrawlDataSchema that stores each recrawl data properties in a separate column in single column
|
||||
* family, whose name may be configured with {@link #setColumnFamily(String)} (default "f").
|
||||
* <ul>
|
||||
* <li>{@code s}: fetch status (as integer text)</li>
|
||||
* <li>{@code d}: content digest (with {@code sha1:} prefix, Base32 text)</li>
|
||||
* <li>{@code e}: ETag (enclosing quotes stripped)</li>
|
||||
* <li>{@code m}: last-modified date-time (as integer timestamp, binary format)</li>
|
||||
* <li>{@code z}: do-not-crawl flag - loader discards URL if this column has non-empty value.</li>
|
||||
* </ul>
|
||||
*
|
||||
* @author kenji
|
||||
*/
|
||||
public class MultiColumnRecrawlDataSchema extends RecrawlDataSchemaBase implements RecrawlDataSchema, RecrawlAttributeConstants {
|
||||
static final Logger logger = Logger.getLogger(MultiColumnRecrawlDataSchema.class.getName());
|
||||
|
||||
public static final byte[] COLUMN_STATUS = Bytes.toBytes("s");
|
||||
public static final byte[] COLUMN_CONTENT_DIGEST = Bytes.toBytes("d");
|
||||
public static final byte[] COLUMN_ETAG = Bytes.toBytes("e");
|
||||
public static final byte[] COLUMN_LAST_MODIFIED = Bytes.toBytes("m");
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.archive.modules.hq.recrawl.RecrawlDataSchema#createPut()
|
||||
*/
|
||||
public Put createPut(CrawlURI uri) {
|
||||
byte[] uriBytes = rowKeyForURI(uri);
|
||||
byte[] key = uriBytes;
|
||||
Put p = new Put(key);
|
||||
String digest = uri.getContentDigestSchemeString();
|
||||
if (digest != null) {
|
||||
p.add(columnFamily, COLUMN_CONTENT_DIGEST, Bytes.toBytes(digest));
|
||||
}
|
||||
p.add(columnFamily, COLUMN_STATUS, Bytes.toBytes(Integer.toString(uri.getFetchStatus())));
|
||||
|
||||
if (uri.isHttpTransaction()) {
|
||||
String etag = uri.getHttpResponseHeader(RecrawlAttributeConstants.A_ETAG_HEADER);
|
||||
if (etag != null) {
|
||||
// Etqg is usually quoted
|
||||
if (etag.length() >= 2 && etag.charAt(0) == '"' && etag.charAt(etag.length() - 1) == '"')
|
||||
etag = etag.substring(1, etag.length() - 1);
|
||||
p.add(columnFamily, COLUMN_ETAG, Bytes.toBytes(etag));
|
||||
}
|
||||
String lastmod = uri.getHttpResponseHeader(RecrawlAttributeConstants.A_LAST_MODIFIED_HEADER);
|
||||
if (lastmod != null) {
|
||||
long lastmod_sec = FetchHistoryHelper.parseHttpDate(lastmod);
|
||||
if (lastmod_sec == 0) {
|
||||
try {
|
||||
lastmod_sec = uri.getFetchCompletedTime();
|
||||
} catch (NullPointerException ex) {
|
||||
logger.warning("CrawlURI.getFetchCompletedTime():" + ex + " for " + uri.shortReportLine());
|
||||
}
|
||||
}
|
||||
if (lastmod_sec != 0)
|
||||
p.add(columnFamily, COLUMN_LAST_MODIFIED, Bytes.toBytes(lastmod_sec));
|
||||
} else {
|
||||
try {
|
||||
long completed = uri.getFetchCompletedTime();
|
||||
if (completed != 0)
|
||||
p.add(columnFamily, COLUMN_LAST_MODIFIED, Bytes.toBytes(completed));
|
||||
} catch (NullPointerException ex) {
|
||||
logger.warning("CrawlURI.getFetchCompletedTime():" + ex + " for " + uri.shortReportLine());
|
||||
}
|
||||
}
|
||||
}
|
||||
return p;
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.archive.modules.hq.recrawl.RecrawlDataSchema#load(java.util.Map, org.apache.hadoop.hbase.client.Result)
|
||||
*/
|
||||
public void load(Result result, CrawlURI curi) {
|
||||
// check for "do-not-crawl" flag - any non-empty data tells not to crawl this
|
||||
// URL.
|
||||
byte[] nocrawl = result.getValue(columnFamily, COLUMN_NOCRAWL);
|
||||
if (nocrawl != null && nocrawl.length > 0) {
|
||||
// fetch status set to S_DEEMED_CHAFF, because this do-not-crawl flag
|
||||
// is primarily intended for preventing crawler from stepping on traps.
|
||||
curi.setFetchStatus(FetchStatusCodes.S_DEEMED_CHAFF);
|
||||
curi.getAnnotations().add("nocrawl");
|
||||
return;
|
||||
}
|
||||
// all column should have identical timestamp.
|
||||
KeyValue rkv = result.getColumnLatest(columnFamily, COLUMN_STATUS);
|
||||
long timestamp = rkv.getTimestamp();
|
||||
Map<String, Object> history = FetchHistoryHelper.getFetchHistory(curi, timestamp, historyLength);
|
||||
// FetchHTTP ignores history with status <= 0
|
||||
byte[] status = result.getValue(columnFamily, COLUMN_STATUS);
|
||||
if (status != null) {
|
||||
// Note that status is stored as integer text. It's typically three-chars
|
||||
// that is less than 4-byte integer bits.
|
||||
history.put(RecrawlAttributeConstants.A_STATUS, Integer.parseInt(Bytes.toString(status)));
|
||||
byte[] etag = result.getValue(columnFamily, COLUMN_ETAG);
|
||||
if (etag != null) {
|
||||
history.put(RecrawlAttributeConstants.A_ETAG_HEADER, Bytes.toString(etag));
|
||||
}
|
||||
byte[] lastmod = result.getValue(columnFamily, COLUMN_LAST_MODIFIED);
|
||||
if (lastmod != null) {
|
||||
long lastmod_sec = Bytes.toLong(lastmod);
|
||||
history.put(RecrawlAttributeConstants.A_LAST_MODIFIED_HEADER, FetchHistoryHelper.formatHttpDate(lastmod_sec));
|
||||
}
|
||||
byte[] digest = result.getValue(columnFamily, COLUMN_CONTENT_DIGEST);
|
||||
if (digest != null) {
|
||||
history.put(RecrawlAttributeConstants.A_CONTENT_DIGEST, Bytes.toString(digest));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,35 +0,0 @@
|
||||
/*
|
||||
* This file is part of the Heritrix web crawler (crawler.archive.org).
|
||||
*
|
||||
* Licensed to the Internet Archive (IA) by one or more individual
|
||||
* contributors.
|
||||
*
|
||||
* The IA 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.archive.modules.recrawl.hbase;
|
||||
|
||||
|
||||
import org.apache.hadoop.hbase.client.Put;
|
||||
import org.apache.hadoop.hbase.client.Result;
|
||||
import org.archive.modules.CrawlURI;
|
||||
|
||||
/**
|
||||
* @author kenji
|
||||
*/
|
||||
public interface RecrawlDataSchema {
|
||||
public String getColumnFamily();
|
||||
public Put createPut(CrawlURI uri);
|
||||
public void load(Result result, CrawlURI curi);
|
||||
// TODO: drop this method by revising createPut(CrawlURI) method.
|
||||
public byte[] rowKeyForURI(CrawlURI curi);
|
||||
}
|
||||
@@ -1,123 +0,0 @@
|
||||
package org.archive.modules.recrawl.hbase;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
import org.apache.hadoop.hbase.util.Bytes;
|
||||
import org.archive.modules.CrawlURI;
|
||||
import org.archive.modules.canonicalize.CanonicalizationRule;
|
||||
import org.archive.modules.recrawl.FetchHistoryHelper;
|
||||
import org.archive.modules.recrawl.FetchHistoryProcessor;
|
||||
import org.archive.modules.recrawl.PersistProcessor;
|
||||
|
||||
/**
|
||||
* implements common utility methods for implementing {@link RecrawlDataSchema}.
|
||||
* <ul>
|
||||
* <li>configuring single column family name</li>
|
||||
* <li>formatting/parsing HTTP date text</li>
|
||||
* <li>constructing row key</li>
|
||||
* <li>preparing fetch-history array</li>
|
||||
* </ul>
|
||||
* @author kenji
|
||||
*/
|
||||
abstract public class RecrawlDataSchemaBase implements RecrawlDataSchema {
|
||||
private static final Logger logger = Logger.getLogger(RecrawlDataSchemaBase.class.getName());
|
||||
|
||||
/**
|
||||
* default value for {@link #columnFamily}.
|
||||
*/
|
||||
public static final byte[] DEFAULT_COLUMN_FAMILY = Bytes.toBytes("f");
|
||||
protected byte[] columnFamily = DEFAULT_COLUMN_FAMILY;
|
||||
|
||||
public static final byte[] COLUMN_NOCRAWL = Bytes.toBytes("z");
|
||||
|
||||
/**
|
||||
* default value for {@link #useCanonicalString}.
|
||||
*/
|
||||
public static boolean DEFAULT_USE_CANONICAL_STRING = true;
|
||||
|
||||
private boolean useCanonicalString = DEFAULT_USE_CANONICAL_STRING;
|
||||
private CanonicalizationRule keyRule = null;
|
||||
|
||||
protected int historyLength = 2;
|
||||
|
||||
public RecrawlDataSchemaBase() {
|
||||
super();
|
||||
}
|
||||
|
||||
public void setColumnFamily(String colf) {
|
||||
columnFamily = Bytes.toBytes(colf);
|
||||
}
|
||||
|
||||
public boolean isUseCanonicalString() {
|
||||
return useCanonicalString;
|
||||
}
|
||||
/**
|
||||
* if set to true, canonicalized string will be used as row key, rather than URI
|
||||
* @param useCanonicalString
|
||||
*/
|
||||
public void setUseCanonicalString(boolean useCanonicalString) {
|
||||
this.useCanonicalString = useCanonicalString;
|
||||
}
|
||||
|
||||
public String getColumnFamily() {
|
||||
return Bytes.toString(columnFamily);
|
||||
}
|
||||
|
||||
|
||||
public CanonicalizationRule getKeyRule() {
|
||||
return keyRule;
|
||||
}
|
||||
/**
|
||||
* alternative canonicalization rule for generating row key from URI.
|
||||
* TODO: currently unused.
|
||||
* @param keyRule
|
||||
*/
|
||||
public void setKeyRule(CanonicalizationRule keyRule) {
|
||||
this.keyRule = keyRule;
|
||||
}
|
||||
|
||||
public int getHistoryLength() {
|
||||
return historyLength;
|
||||
}
|
||||
|
||||
/**
|
||||
* maximum number of crawl history entries to retain in {@link CrawlURI}.
|
||||
* when more than this number of crawl history entry is being added by
|
||||
* {@link #getFetchHistory(CrawlURI, long)}, oldest entry will be discarded.
|
||||
* {@code historyLength} should be the same number as
|
||||
* {@link FetchHistoryProcessor#setHistoryLength(int)}, or FetchHistoryProcessor will
|
||||
* reallocate the crawl history array.
|
||||
* @param historyLength
|
||||
* @see FetchHistoryProcessor#setHistoryLength(int)
|
||||
*/
|
||||
public void setHistoryLength(int historyLength) {
|
||||
this.historyLength = historyLength;
|
||||
}
|
||||
|
||||
/**
|
||||
* calls {@link FetchHistoryHelper#getFetchHistory(CrawlURI, long, int)} with {@link #historyLength}.
|
||||
* @param uri CrawlURI from which fetch history is obtained.
|
||||
* @return Map object for storing re-crawl data (never null).
|
||||
* @see FetchHistoryHelper#getFetchHistory(CrawlURI, long, int)
|
||||
* @see FetchHistoryProcessor
|
||||
*/
|
||||
protected Map<String, Object> getFetchHistory(CrawlURI uri, long timestamp) {
|
||||
return FetchHistoryHelper.getFetchHistory(uri, timestamp, historyLength);
|
||||
}
|
||||
|
||||
/**
|
||||
* return row key for {@code curi}.
|
||||
* TODO: move this to HBasePersistProcessor by redesigning {@link RecrawlDataSchema}.
|
||||
* @param curi {@link CrawlURI} for which a row is being fetched.
|
||||
* @return row key
|
||||
*/
|
||||
public byte[] rowKeyForURI(CrawlURI curi) {
|
||||
if (useCanonicalString) {
|
||||
// TODO: use keyRule if specified.
|
||||
return Bytes.toBytes(PersistProcessor.persistKeyFor(curi));
|
||||
} else {
|
||||
return Bytes.toBytes(curi.toString());
|
||||
}
|
||||
}
|
||||
}
|
||||
-174
@@ -1,174 +0,0 @@
|
||||
/*
|
||||
* This file is part of the Heritrix web crawler (crawler.archive.org).
|
||||
*
|
||||
* Licensed to the Internet Archive (IA) by one or more individual
|
||||
* contributors.
|
||||
*
|
||||
* The IA 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.archive.modules.recrawl.hbase;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.logging.Level;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
import org.apache.commons.httpclient.HttpMethod;
|
||||
import org.apache.hadoop.hbase.KeyValue;
|
||||
import org.apache.hadoop.hbase.client.Put;
|
||||
import org.apache.hadoop.hbase.client.Result;
|
||||
import org.apache.hadoop.hbase.util.Bytes;
|
||||
import org.archive.modules.CrawlURI;
|
||||
import org.archive.modules.fetcher.FetchStatusCodes;
|
||||
import org.archive.modules.recrawl.FetchHistoryHelper;
|
||||
import org.archive.modules.recrawl.RecrawlAttributeConstants;
|
||||
import org.json.JSONException;
|
||||
import org.json.JSONObject;
|
||||
|
||||
/**
|
||||
* {@linkplain SingleColumnJsonRecrawlDataSchema} stores all re-crawl data properties in a single column,
|
||||
* in JSON format. As HBase stores each column paired with the row key, it takes a lot of space to store
|
||||
* each re-crawl data property in its own column.
|
||||
* <ul>
|
||||
* <li>{@code r}: re-crawl data in JSON format</li>
|
||||
* <li>{@code z}: do-not-crawl flag - loader discards URL if this column has non-empty value.</li>
|
||||
* </ul>
|
||||
* @author Kenji Nagahashi
|
||||
*/
|
||||
public class SingleColumnJsonRecrawlDataSchema extends RecrawlDataSchemaBase
|
||||
implements RecrawlDataSchema {
|
||||
static final Logger logger = Logger.getLogger(SingleColumnJsonRecrawlDataSchema.class.getName());
|
||||
|
||||
public static byte[] DEFAULT_COLUMN = Bytes.toBytes("r");
|
||||
|
||||
// JSON property names for re-crawl data properties
|
||||
public static final String PROPERTY_STATUS = "s";
|
||||
public static final String PROPERTY_CONTENT_DIGEST = "d";
|
||||
public static final String PROPERTY_ETAG = "e";
|
||||
public static final String PROPERTY_LAST_MODIFIED = "m";
|
||||
|
||||
// SHA1 scheme is assumed.
|
||||
public static final String CONTENT_DIGEST_SCHEME = "sha1:";
|
||||
|
||||
// single column for storing JSON of re-crawl data
|
||||
protected byte[] column = DEFAULT_COLUMN;
|
||||
public void setColumn(String column) {
|
||||
this.column = Bytes.toBytes(column);
|
||||
}
|
||||
public String getColumn() {
|
||||
return Bytes.toString(column);
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.archive.modules.hq.recrawl.RecrawlDataSchema#createPut(org.archive.modules.CrawlURI)
|
||||
*/
|
||||
public Put createPut(CrawlURI uri) {
|
||||
byte[] key = rowKeyForURI(uri);
|
||||
Put p = new Put(key);
|
||||
JSONObject jo = new JSONObject();
|
||||
try {
|
||||
// TODO should we post warning message when scheme != "sha1"?
|
||||
String digest = uri.getContentDigestString();
|
||||
if (digest != null) {
|
||||
jo.put(PROPERTY_CONTENT_DIGEST, digest);
|
||||
}
|
||||
jo.put(PROPERTY_STATUS, uri.getFetchStatus());
|
||||
if (uri.isHttpTransaction()) {
|
||||
String etag = uri.getHttpResponseHeader(RecrawlAttributeConstants.A_ETAG_HEADER);
|
||||
if (etag != null) {
|
||||
// Etag is usually quoted
|
||||
if (etag.length() >= 2 && etag.charAt(0) == '"' && etag.charAt(etag.length() - 1) == '"')
|
||||
etag = etag.substring(1, etag.length() - 1);
|
||||
jo.put(PROPERTY_ETAG, etag);
|
||||
}
|
||||
String lastmod = uri.getHttpResponseHeader(RecrawlAttributeConstants.A_LAST_MODIFIED_HEADER);
|
||||
if (lastmod != null) {
|
||||
long lastmod_sec = FetchHistoryHelper.parseHttpDate(lastmod);
|
||||
if (lastmod_sec == 0) {
|
||||
try {
|
||||
lastmod_sec = uri.getFetchCompletedTime();
|
||||
} catch (NullPointerException ex) {
|
||||
logger.warning("CrawlURI.getFetchCompletedTime():" + ex + " for " + uri.shortReportLine());
|
||||
}
|
||||
}
|
||||
} else {
|
||||
try {
|
||||
long completed = uri.getFetchCompletedTime();
|
||||
if (completed != 0)
|
||||
jo.put(PROPERTY_LAST_MODIFIED, completed);
|
||||
} catch (NullPointerException ex) {
|
||||
logger.warning("CrawlURI.getFetchCompletedTime():" + ex + " for " + uri.shortReportLine());
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (JSONException ex) {
|
||||
// should not happen - all values are either primitive or String.
|
||||
logger.log(Level.SEVERE, "JSON translation failed", ex);
|
||||
}
|
||||
p.add(columnFamily, column, Bytes.toBytes(jo.toString()));
|
||||
return p;
|
||||
}
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.archive.modules.hq.recrawl.RecrawlDataSchema#load(org.apache.hadoop.hbase.client.Result)
|
||||
*/
|
||||
public void load(Result result, CrawlURI curi) {
|
||||
// check for "do-not-crawl" flag - any non-empty data tells not to crawl this
|
||||
// URL.
|
||||
byte[] nocrawl = result.getValue(columnFamily, COLUMN_NOCRAWL);
|
||||
if (nocrawl != null && nocrawl.length > 0) {
|
||||
// fetch status set to S_DEEMED_CHAFF, because this do-not-crawl flag
|
||||
// is primarily intended for preventing crawler from stepping on traps.
|
||||
curi.setFetchStatus(FetchStatusCodes.S_DEEMED_CHAFF);
|
||||
curi.getAnnotations().add("nocrawl");
|
||||
return;
|
||||
}
|
||||
|
||||
KeyValue rkv = result.getColumnLatest(columnFamily, column);
|
||||
long timestamp = rkv.getTimestamp();
|
||||
Map<String, Object> history = FetchHistoryHelper.getFetchHistory(curi, timestamp, historyLength);
|
||||
if (history == null) {
|
||||
// crawl history array is fully occupied by crawl history entries
|
||||
// newer than timestamp.
|
||||
return;
|
||||
}
|
||||
byte[] jsonBytes = rkv.getValue();
|
||||
if (jsonBytes != null) {
|
||||
JSONObject jo = null;
|
||||
try {
|
||||
jo = new JSONObject(Bytes.toString(jsonBytes));
|
||||
} catch (JSONException ex) {
|
||||
logger.warning(String.format("JSON parsing failed for key %1s: %2s",
|
||||
result.getRow(), ex.getMessage()));
|
||||
}
|
||||
if (jo != null) {
|
||||
int status = jo.optInt(PROPERTY_STATUS, -1);
|
||||
if (status >= 0) {
|
||||
history.put(RecrawlAttributeConstants.A_STATUS, status);
|
||||
}
|
||||
String digest = jo.optString(PROPERTY_CONTENT_DIGEST);
|
||||
if (digest != null) {
|
||||
history.put(RecrawlAttributeConstants.A_CONTENT_DIGEST, CONTENT_DIGEST_SCHEME + digest);
|
||||
}
|
||||
String etag = jo.optString(PROPERTY_ETAG);
|
||||
if (etag != null) {
|
||||
history.put(RecrawlAttributeConstants.A_ETAG_HEADER, etag);
|
||||
}
|
||||
long lastmod = jo.optLong(PROPERTY_LAST_MODIFIED);
|
||||
if (lastmod > 0) {
|
||||
history.put(RecrawlAttributeConstants.A_LAST_MODIFIED_HEADER, FetchHistoryHelper.formatHttpDate(lastmod));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,380 +0,0 @@
|
||||
/*
|
||||
* This file is part of the Heritrix web crawler (crawler.archive.org).
|
||||
*
|
||||
* Licensed to the Internet Archive (IA) by one or more individual
|
||||
* contributors.
|
||||
*
|
||||
* The IA 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.archive.modules.recrawl.hbase;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.concurrent.locks.Lock;
|
||||
import java.util.concurrent.locks.ReentrantReadWriteLock;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.apache.hadoop.hbase.HTableDescriptor;
|
||||
import org.apache.hadoop.hbase.NotServingRegionException;
|
||||
import org.apache.hadoop.hbase.TableNotFoundException;
|
||||
import org.apache.hadoop.hbase.client.Get;
|
||||
import org.apache.hadoop.hbase.client.HTable;
|
||||
import org.apache.hadoop.hbase.client.HTableInterface;
|
||||
import org.apache.hadoop.hbase.client.Put;
|
||||
import org.apache.hadoop.hbase.client.Result;
|
||||
import org.apache.hadoop.hbase.util.Bytes;
|
||||
|
||||
/**
|
||||
* simple HTable wrapper that shares single instance of HTable among threads.
|
||||
* If you only perform get on HTable, this implementation
|
||||
* should be good enough. If multiple threads performs Put, {@link HBaseTable}
|
||||
* would be more efficient.
|
||||
* <p>when HBase I/O fails due to issue with network/region server/zookeeper, this
|
||||
* class waits for preset time (see {@link #setReconnectInterval(int)})
|
||||
* before trying to reestablish HBase connection. During this hold-ff period, all
|
||||
* {@link #get(Get)} and {@link #put(Put)} calls will fail.
|
||||
*
|
||||
* @author kenji
|
||||
*/
|
||||
public class SingleHBaseTable extends HBaseTableBean {
|
||||
private static final Log LOG = LogFactory.getLog(SingleHBaseTable.class);
|
||||
|
||||
private HTableInterface table;
|
||||
private volatile long tableError;
|
||||
private ReentrantReadWriteLock tableUseLock = new ReentrantReadWriteLock();
|
||||
|
||||
boolean autoReconnect = true;
|
||||
|
||||
public boolean isAutoReconnect() {
|
||||
return autoReconnect;
|
||||
}
|
||||
/**
|
||||
* if set to {@code true}, HBaseClient tries to reconnect to the HBase master
|
||||
* immediately when Put request failed due to connection loss (note {@link #put(Put)}
|
||||
* still throws IOException even if autoReconnect is enabled.)
|
||||
* @param autoReconnect true to enable auto-reconnect
|
||||
*/
|
||||
public void setAutoReconnect(boolean autoReconnect) {
|
||||
this.autoReconnect = autoReconnect;
|
||||
}
|
||||
|
||||
protected boolean autoFlush = true;
|
||||
/**
|
||||
* passed on to HTable's autoFlush property upon creation.
|
||||
* @return true for enabling auto-flush.
|
||||
*/
|
||||
public boolean isAutoFlush() {
|
||||
return autoFlush;
|
||||
}
|
||||
public void setAutoFlush(boolean autoFlush) {
|
||||
this.autoFlush = autoFlush;
|
||||
}
|
||||
|
||||
// default 3 minutes
|
||||
private int reconnectInterval = 1000 * 3 * 60;
|
||||
|
||||
public int getReconnectInterval() {
|
||||
return reconnectInterval;
|
||||
}
|
||||
/**
|
||||
* set hold-off interval upon communication errors.
|
||||
* @param reconnectInterval hold-off interval in milliseconds.
|
||||
*/
|
||||
public void setReconnectInterval(int reconnectInterval) {
|
||||
this.reconnectInterval = reconnectInterval;
|
||||
}
|
||||
|
||||
// counters
|
||||
|
||||
protected AtomicLong getCount = new AtomicLong();
|
||||
// count of GET/PUT failures (i.e. not counting connection failures).
|
||||
protected AtomicLong getErrorCount = new AtomicLong();
|
||||
protected AtomicLong getSkipCount = new AtomicLong();
|
||||
|
||||
protected AtomicLong putCount = new AtomicLong();
|
||||
protected AtomicLong putErrorCount = new AtomicLong();
|
||||
protected AtomicLong putSkipCount = new AtomicLong();
|
||||
|
||||
protected AtomicLong connectCount = new AtomicLong();
|
||||
|
||||
public long getGetCount() { return getCount.get(); }
|
||||
public long getGetErrorCount() { return getErrorCount.get(); }
|
||||
public long getGetSkipCount() { return getSkipCount.get(); }
|
||||
public long getPutCount() { return putCount.get(); }
|
||||
public long getConnectCount() { return connectCount.get(); }
|
||||
|
||||
// for diagnosing deadlock situation
|
||||
public Map<String, Object> getTableLockState() {
|
||||
Map<String, Object> m = new LinkedHashMap<String, Object>();
|
||||
m.put("readLockCount", tableUseLock.getReadLockCount());
|
||||
m.put("queueLength", tableUseLock.getQueueLength());
|
||||
m.put("writeLocked", tableUseLock.isWriteLocked());
|
||||
return m;
|
||||
}
|
||||
|
||||
public SingleHBaseTable() {
|
||||
}
|
||||
|
||||
/**
|
||||
* attempts to reconnect to HBase if table is null.
|
||||
* must not be called with read-lock.
|
||||
* @return existing or newly opened HTableInterface.
|
||||
*/
|
||||
protected HTableInterface getTable() {
|
||||
if (table == null && autoReconnect)
|
||||
openTable();
|
||||
return table;
|
||||
}
|
||||
/**
|
||||
* close HTable {@code table}, set current time to tableError if closing because
|
||||
* of a communication error. should be called with write lock.
|
||||
* @param htable HTable to close.
|
||||
* @param byError true if closing because of an error.
|
||||
*/
|
||||
protected void closeTable(HTableInterface htable, boolean byError) {
|
||||
if (htable == null) return;
|
||||
if (table != htable) {
|
||||
// other thread did closeTable on htable. don't close table.
|
||||
return;
|
||||
}
|
||||
try {
|
||||
table = null;
|
||||
htable.close();
|
||||
} catch (IOException ex) {
|
||||
LOG.warn("error closing " + htable + " - some commits may have been lost");
|
||||
}
|
||||
if (byError) {
|
||||
tableError = System.currentTimeMillis();
|
||||
}
|
||||
}
|
||||
|
||||
public void put(Put p) throws IOException {
|
||||
putCount.incrementAndGet();
|
||||
// trigger reconnection if necessary. as table can be modified before
|
||||
// read lock is acquired, we don't read table variable here.
|
||||
getTable();
|
||||
boolean htableFailed = false;
|
||||
HTableInterface htable = null;
|
||||
Lock readLock = tableUseLock.readLock();
|
||||
try {
|
||||
if (!readLock.tryLock(TRY_READ_LOCK_TIMEOUT, TimeUnit.SECONDS)) {
|
||||
putSkipCount.incrementAndGet();
|
||||
throw new IOException("could not acquire read lock for HTable.");
|
||||
}
|
||||
} catch (InterruptedException ex) {
|
||||
throw new IOException("interrupted while acquiring read lock", ex);
|
||||
}
|
||||
try {
|
||||
htable = table;
|
||||
if (htable == null) {
|
||||
putSkipCount.incrementAndGet();
|
||||
throw new IOException("HBase connection is unvailable.");
|
||||
}
|
||||
// HTable.put() buffers Puts and access to the buffer is not
|
||||
// synchronized.
|
||||
synchronized (htable) {
|
||||
try {
|
||||
htable.put(p);
|
||||
} catch (NullPointerException ex) {
|
||||
// HTable.put() throws NullPointerException when connection is lost.
|
||||
// It is somewhat weird, so translate it to IOException.
|
||||
putErrorCount.incrementAndGet();
|
||||
htableFailed = true;
|
||||
throw new IOException("hbase connection is lost", ex);
|
||||
} catch (NotServingRegionException ex) {
|
||||
putErrorCount.incrementAndGet();
|
||||
// no need to close HTable.
|
||||
throw ex;
|
||||
} catch (IOException ex) {
|
||||
putErrorCount.incrementAndGet();
|
||||
htableFailed = true;
|
||||
throw ex;
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
readLock.unlock();
|
||||
if (htableFailed) {
|
||||
closeTable(htable, true);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public Result get(Get g) throws IOException {
|
||||
getCount.incrementAndGet();
|
||||
// trigger reconnection if necessary. as table can be modified before
|
||||
// read lock is acquired, we don't read table variable here.
|
||||
getTable();
|
||||
boolean htableFailed = false;
|
||||
HTableInterface htable = null;
|
||||
Lock readLock = tableUseLock.readLock();
|
||||
try {
|
||||
if (!readLock.tryLock(TRY_READ_LOCK_TIMEOUT, TimeUnit.SECONDS)) {
|
||||
getSkipCount.incrementAndGet();
|
||||
throw new IOException("could not acquire read lock for HTable.");
|
||||
}
|
||||
} catch (InterruptedException ex) {
|
||||
throw new IOException("interrupted while acquiring read lock", ex);
|
||||
}
|
||||
try {
|
||||
htable = table;
|
||||
if (htable == null) {
|
||||
getSkipCount.incrementAndGet();
|
||||
throw new IOException("HBase connection is unvailable.");
|
||||
}
|
||||
try {
|
||||
return htable.get(g);
|
||||
} catch (NotServingRegionException ex) {
|
||||
// caused by disruption to HBase cluster. no need to
|
||||
// refresh HBase connection, since connection itself
|
||||
// is working okay.
|
||||
// TODO: should we need to back-off for a while? other
|
||||
// regions may still be accessible.
|
||||
getErrorCount.incrementAndGet();
|
||||
throw ex;
|
||||
} catch (IOException ex) {
|
||||
getErrorCount.incrementAndGet();
|
||||
htableFailed = true;
|
||||
throw ex;
|
||||
}
|
||||
} finally {
|
||||
readLock.unlock();
|
||||
if (htableFailed) {
|
||||
closeTable(htable, true);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public HTableDescriptor getHtableDescriptor() throws IOException {
|
||||
HTableInterface table = getTable();
|
||||
if (table == null) {
|
||||
throw new IOException("HBase connection is unavailable.");
|
||||
}
|
||||
return table.getTableDescriptor();
|
||||
}
|
||||
|
||||
public boolean inBackoffPeriod() {
|
||||
return (tableError > 0 &&
|
||||
(System.currentTimeMillis() - tableError) < reconnectInterval);
|
||||
}
|
||||
|
||||
/**
|
||||
* timestamp of the last Put/Get error.
|
||||
* @return timestamp in ms.
|
||||
*/
|
||||
public long getTableErrorTime() {
|
||||
return tableError;
|
||||
}
|
||||
/**
|
||||
* connect to HBase.
|
||||
* it does nothing if table is non-null, or it is in the back-off period since
|
||||
* the last error.
|
||||
* should be called with write lock.
|
||||
*/
|
||||
protected boolean openTable() {
|
||||
if (table != null) return true;
|
||||
// fail immediately if we're in back-off period.
|
||||
if (inBackoffPeriod()) return false;
|
||||
try {
|
||||
HTable t = new HTable(hbase.configuration(), Bytes.toBytes(htableName));
|
||||
connectCount.incrementAndGet();
|
||||
t.setAutoFlush(autoFlush);
|
||||
table = t;
|
||||
tableError = 0;
|
||||
return true;
|
||||
} catch (TableNotFoundException ex) {
|
||||
// ex.getMessage() only has table name. be a little bit more friendly.
|
||||
LOG.warn("failed to connect to HTable \"" + htableName + "\": Table Not Found");
|
||||
tableError = System.currentTimeMillis();
|
||||
return false;
|
||||
} catch (IOException ex) {
|
||||
LOG.warn("failed to connect to HTable \"" + htableName + "\" (" + ex.getMessage() + ")");
|
||||
tableError = System.currentTimeMillis();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
/**
|
||||
* number of seconds to wait for acquiring read lock.
|
||||
* if read lock is not acquired within this many seconds (probably
|
||||
* due to deadlock situation on write-lock side), {@link #get(Get)} will
|
||||
* <i>silently</i> fail.
|
||||
*/
|
||||
public final static long TRY_READ_LOCK_TIMEOUT = 5;
|
||||
/**
|
||||
* number of seconds to wait for acquiring write lock.
|
||||
*/
|
||||
public final static long TRY_WRITE_LOCK_TIMEOUT = 10;
|
||||
|
||||
/**
|
||||
* close current connection and establish new connection.
|
||||
* fails silently if back-off period is in effect.
|
||||
*/
|
||||
protected void reconnect(boolean onerror) throws IOException, InterruptedException {
|
||||
// avoid deadlock situation caused by attempting
|
||||
// to acquire write lock while holding read lock.
|
||||
// there'd be no real dead-lock now that timeout on write lock is implemented,
|
||||
// but it's nice to know there's a bug in locking.
|
||||
if (tableUseLock.getReadHoldCount() > 0) {
|
||||
LOG.warn("avoiding deadlock: reconnect() called by thread with read lock.");
|
||||
return;
|
||||
}
|
||||
Lock writeLock = tableUseLock.writeLock();
|
||||
if (!writeLock.tryLock(TRY_WRITE_LOCK_TIMEOUT, TimeUnit.SECONDS)) {
|
||||
LOG.warn("reconnect() could not acquire write lock on tableUseLock for " +
|
||||
TRY_WRITE_LOCK_TIMEOUT + "s, giving up.");
|
||||
return;
|
||||
}
|
||||
try {
|
||||
closeTable(table, onerror);
|
||||
openTable();
|
||||
} finally {
|
||||
writeLock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* close current connection and establish new connection.
|
||||
* for refreshing stale connection through scripting.
|
||||
* resets tableErrorTime to zero (it will be set to non-zero if
|
||||
* reconnection attempt fails).
|
||||
* @throws IOException
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
public void reconnect() throws IOException, InterruptedException {
|
||||
tableError = 0;
|
||||
reconnect(false);
|
||||
}
|
||||
|
||||
// public boolean isRunning() {
|
||||
// return table != null;
|
||||
// }
|
||||
public void start() {
|
||||
super.start();
|
||||
openTable();
|
||||
}
|
||||
public void stop() {
|
||||
if (table != null) {
|
||||
try {
|
||||
table.close();
|
||||
} catch (IOException ex) {
|
||||
LOG.warn("table.close() failed", ex);
|
||||
}
|
||||
}
|
||||
table = null;
|
||||
super.stop();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user