From 57a2c00ccf7ea2960456cb6d0a49022a722a1158 Mon Sep 17 00:00:00 2001 From: John Blakeney Date: Tue, 1 Sep 2015 22:00:58 -0400 Subject: Fix various CDH5 integration problems. --- pom.xml | 15 +++++++++++- .../spyglass/hbase/HBaseInputFormatGranular.java | 27 ++++++++++++++-------- .../parallelai/spyglass/hbase/HBaseRawTap.java | 2 +- .../spyglass/hbase/HBaseRecordReaderGranular.java | 11 ++++----- .../java/parallelai/spyglass/hbase/HBaseTap.java | 2 +- .../hbase/example/HBaseExampleRunner.scala | 1 - 6 files changed, 39 insertions(+), 19 deletions(-) diff --git a/pom.xml b/pom.xml index 861a61d..90261ac 100644 --- a/pom.xml +++ b/pom.xml @@ -35,6 +35,8 @@ 2.5.0-mr1-${cdh.version} 0.98.6-${cdh.version} + 2.8.2 + 2.10.3 @@ -153,7 +155,12 @@ org.apache.hbase - hbase + hbase-client + ${hbase.version} + + + org.apache.hbase + hbase-server ${hbase.version} @@ -169,6 +176,12 @@ ${typesafe.config.version} + + joda-time + joda-time + ${joda-time.version} + + com.twitter.elephantbird elephant-bird-core diff --git a/src/main/java/parallelai/spyglass/hbase/HBaseInputFormatGranular.java b/src/main/java/parallelai/spyglass/hbase/HBaseInputFormatGranular.java index 332bbd7..ae9db71 100644 --- a/src/main/java/parallelai/spyglass/hbase/HBaseInputFormatGranular.java +++ b/src/main/java/parallelai/spyglass/hbase/HBaseInputFormatGranular.java @@ -2,6 +2,8 @@ package parallelai.spyglass.hbase; import java.io.IOException; import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.UnknownHostException; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -14,7 +16,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.HRegionLocation; -import org.apache.hadoop.hbase.HServerAddress; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.io.ImmutableBytesWritable; import org.apache.hadoop.hbase.util.Addressing; @@ -110,15 +111,15 @@ public class HBaseInputFormatGranular extends HBaseInputFormatBase { : maxKey; HRegionLocation regionLoc = table.getRegionLocation(keys.getFirst()[i]); - HServerAddress regionServerAddress = regionLoc.getServerAddress(); - InetAddress regionAddress = regionServerAddress.getInetSocketAddress().getAddress(); + String regionServerHostnamePort = regionLoc.getHostnamePort(); + InetAddress regionAddress = toInetAddress(regionServerHostnamePort); String regionLocation; try { regionLocation = reverseDNS(regionAddress); } catch (NamingException e) { LOG.error("Cannot resolve the host name for " + regionAddress + " because of " + e); - regionLocation = regionServerAddress.getHostname(); + regionLocation = toHostname(regionServerHostnamePort); } regionNames[i] = regionLoc.getRegionInfo().getRegionNameAsString(); @@ -189,17 +190,17 @@ public class HBaseInputFormatGranular extends HBaseInputFormatBase { byte[] rStart = cRegion.getRegionInfo().getStartKey(); byte[] rStop = cRegion.getRegionInfo().getEndKey(); - HServerAddress regionServerAddress = cRegion - .getServerAddress(); - InetAddress regionAddress = regionServerAddress - .getInetSocketAddress().getAddress(); + String regionServerHostnamePort = cRegion + .getHostnamePort(); + InetAddress regionAddress = + toInetAddress(regionServerHostnamePort); String regionLocation; try { regionLocation = reverseDNS(regionAddress); } catch (NamingException e) { LOG.error("Cannot resolve the host name for " + regionAddress + " because of " + e); - regionLocation = regionServerAddress.getHostname(); + regionLocation = toHostname(regionServerHostnamePort); } String regionName = cRegion.getRegionInfo().getRegionNameAsString(); @@ -361,6 +362,14 @@ public class HBaseInputFormatGranular extends HBaseInputFormatBase { } } + private String toHostname(String regionServerHostnamePort) { + return Addressing.parseHostname(regionServerHostnamePort); + } + + private InetAddress toInetAddress(String regionServerHostnamePort) throws UnknownHostException { + return InetAddress.getByName(toHostname(regionServerHostnamePort)); + } + private String reverseDNS(InetAddress ipAddress) throws NamingException { String hostName = this.reverseDNSCacheMap.get(ipAddress); if (hostName == null) { diff --git a/src/main/java/parallelai/spyglass/hbase/HBaseRawTap.java b/src/main/java/parallelai/spyglass/hbase/HBaseRawTap.java index efe548d..ece7e61 100644 --- a/src/main/java/parallelai/spyglass/hbase/HBaseRawTap.java +++ b/src/main/java/parallelai/spyglass/hbase/HBaseRawTap.java @@ -156,7 +156,7 @@ public class HBaseRawTap extends Tap { return new Path(SCHEME + ":/" + tableName.replaceAll(":", "_")); } - protected HBaseAdmin getHBaseAdmin(JobConf conf) throws MasterNotRunningException, ZooKeeperConnectionException { + protected HBaseAdmin getHBaseAdmin(JobConf conf) throws IOException { if (hBaseAdmin == null) { Configuration hbaseConf = HBaseConfiguration.create(conf); hBaseAdmin = new HBaseAdmin(hbaseConf); diff --git a/src/main/java/parallelai/spyglass/hbase/HBaseRecordReaderGranular.java b/src/main/java/parallelai/spyglass/hbase/HBaseRecordReaderGranular.java index b34ed3f..8a9559f 100644 --- a/src/main/java/parallelai/spyglass/hbase/HBaseRecordReaderGranular.java +++ b/src/main/java/parallelai/spyglass/hbase/HBaseRecordReaderGranular.java @@ -10,7 +10,6 @@ import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.io.ImmutableBytesWritable; import org.apache.hadoop.hbase.mapreduce.TableInputFormat; import org.apache.hadoop.hbase.util.Bytes; -import org.apache.hadoop.hbase.util.Writables; import org.apache.hadoop.util.StringUtils; import java.io.IOException; @@ -251,7 +250,7 @@ public class HBaseRecordReaderGranular extends HBaseRecordReaderBase { } lastSuccessfulRow = key.get(); - Writables.copyWritable(result, value); + value.copyFrom(result); return true; } return false; @@ -302,8 +301,8 @@ public class HBaseRecordReaderGranular extends HBaseRecordReaderBase { key.set(result.getRow()); } lastSuccessfulRow = key.get(); - Writables.copyWritable(result, value); - + value.copyFrom(result); + return true; } else { LOG.debug(" Key ("+ Bytes.toString(nextKey)+ ") return an EMPTY result. Get ("+theGet.getId()+")" ); //alg0 @@ -352,7 +351,7 @@ public class HBaseRecordReaderGranular extends HBaseRecordReaderBase { key.set(result.getRow()); } lastSuccessfulRow = key.get(); - Writables.copyWritable(result, value); + value.copyFrom(result); return true; } else { @@ -408,7 +407,7 @@ public class HBaseRecordReaderGranular extends HBaseRecordReaderBase { key.set(result.getRow()); } lastSuccessfulRow = key.get(); - Writables.copyWritable(result, value); + value.copyFrom(result); return true; } else { LOG.debug(String.format("+ Key (%s) return an EMPTY result. Get (%s)", Bytes.toString(nextKey), theGet.getId()) ); //alg0 diff --git a/src/main/java/parallelai/spyglass/hbase/HBaseTap.java b/src/main/java/parallelai/spyglass/hbase/HBaseTap.java index e4a3f78..e1897ef 100644 --- a/src/main/java/parallelai/spyglass/hbase/HBaseTap.java +++ b/src/main/java/parallelai/spyglass/hbase/HBaseTap.java @@ -143,7 +143,7 @@ public class HBaseTap extends Tap { return new Path(SCHEME + ":/" + tableName.replaceAll(":", "_")); } - protected HBaseAdmin getHBaseAdmin(JobConf conf) throws MasterNotRunningException, ZooKeeperConnectionException { + protected HBaseAdmin getHBaseAdmin(JobConf conf) throws IOException { if (hBaseAdmin == null) { Configuration hbaseConf = HBaseConfiguration.create(conf); hBaseAdmin = new HBaseAdmin(hbaseConf); diff --git a/src/main/scala/parallelai/spyglass/hbase/example/HBaseExampleRunner.scala b/src/main/scala/parallelai/spyglass/hbase/example/HBaseExampleRunner.scala index c503247..949b86e 100644 --- a/src/main/scala/parallelai/spyglass/hbase/example/HBaseExampleRunner.scala +++ b/src/main/scala/parallelai/spyglass/hbase/example/HBaseExampleRunner.scala @@ -7,7 +7,6 @@ import parallelai.spyglass.base.JobRunner import org.apache.hadoop.conf.Configuration import org.apache.hadoop.hbase.{HColumnDescriptor, HTableDescriptor, HBaseConfiguration} import org.apache.hadoop.hbase.client.{Put, HTable, HConnectionManager, HBaseAdmin} -import org.apache.hadoop.hbase.io.hfile.Compression import org.apache.hadoop.hbase.regionserver.StoreFile import org.apache.hadoop.hbase.util.Bytes import parallelai.spyglass.hbase.HBaseSalter -- cgit v1.2.3 From ead7e97dca4023fa7a7a3337a2ac1d238f6e64c4 Mon Sep 17 00:00:00 2001 From: John Blakeney Date: Thu, 3 Sep 2015 20:12:21 -0400 Subject: Increment version. Fix Maven plugin versions. --- pom.xml | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 90261ac..d22d4c4 100644 --- a/pom.xml +++ b/pom.xml @@ -11,7 +11,7 @@ Cascading and Scalding wrapper for HBase with advanced features parallelai parallelai.spyglass - 2.10_0.12.0_5.3.0 + 2.10_0.12.1_5.3.0 jar @@ -27,6 +27,7 @@ 3.0 2.15.2 + 2.4 2.12.3 cdh5.3.0 @@ -233,6 +234,7 @@ org.apache.maven.plugins maven-source-plugin + ${maven-source-plugin.version} attach-sources @@ -246,6 +248,7 @@ org.scala-tools maven-scala-plugin + ${maven-scala-plugin.version} -- cgit v1.2.3