Added a HdfsResource implementation Project: http://git-wip-us.apache.org/repos/asf/metamodel/repo Commit: http://git-wip-us.apache.org/repos/asf/metamodel/commit/9191b9fb Tree: http://git-wip-us.apache.org/repos/asf/metamodel/tree/9191b9fb Diff: http://git-wip-us.apache.org/repos/asf/metamodel/diff/9191b9fb
Branch: refs/heads/master Commit: 9191b9fbec1c36e4eea5a790d1ac77bbe6284623 Parents: e8f0624 Author: Kasper Sørensen <[email protected]> Authored: Fri Jun 5 14:03:21 2015 +0200 Committer: Kasper Sørensen <[email protected]> Committed: Fri Jun 5 14:03:21 2015 +0200 ---------------------------------------------------------------------- .../java/org/apache/metamodel/util/Func.java | 16 +- .../apache/metamodel/util/UncheckedFunc.java | 36 +++ .../org/apache/metamodel/util/HdfsResource.java | 265 +++++++++++++++++++ .../apache/metamodel/util/HdfsResourceTest.java | 28 ++ .../metamodel/hbase/HBaseDataContext.java | 1 + 5 files changed, 338 insertions(+), 8 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/metamodel/blob/9191b9fb/core/src/main/java/org/apache/metamodel/util/Func.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/metamodel/util/Func.java b/core/src/main/java/org/apache/metamodel/util/Func.java index 684411f..44014e0 100644 --- a/core/src/main/java/org/apache/metamodel/util/Func.java +++ b/core/src/main/java/org/apache/metamodel/util/Func.java @@ -30,12 +30,12 @@ package org.apache.metamodel.util; */ public interface Func<I, O> { - /** - * Evaluates an element and transforms it using this function. - * - * @param arg - * the input given to the function - * @return the output result of the function - */ - public O eval(I arg); + /** + * Evaluates an element and transforms it using this function. + * + * @param arg + * the input given to the function + * @return the output result of the function + */ + public O eval(I arg); } http://git-wip-us.apache.org/repos/asf/metamodel/blob/9191b9fb/core/src/main/java/org/apache/metamodel/util/UncheckedFunc.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/metamodel/util/UncheckedFunc.java b/core/src/main/java/org/apache/metamodel/util/UncheckedFunc.java new file mode 100644 index 0000000..ed5d89c --- /dev/null +++ b/core/src/main/java/org/apache/metamodel/util/UncheckedFunc.java @@ -0,0 +1,36 @@ +/** + * 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.metamodel.util; + +public abstract class UncheckedFunc<I, O> implements Func<I, O> { + + @Override + public O eval(I arg) { + try { + return evalUnchecked(arg); + } catch (Exception e) { + if (e instanceof RuntimeException) { + throw (RuntimeException) e; + } + throw new RuntimeException(e); + } + } + + protected abstract O evalUnchecked(I arg) throws Exception; +} http://git-wip-us.apache.org/repos/asf/metamodel/blob/9191b9fb/hadoop/src/main/java/org/apache/metamodel/util/HdfsResource.java ---------------------------------------------------------------------- diff --git a/hadoop/src/main/java/org/apache/metamodel/util/HdfsResource.java b/hadoop/src/main/java/org/apache/metamodel/util/HdfsResource.java new file mode 100644 index 0000000..bd98062 --- /dev/null +++ b/hadoop/src/main/java/org/apache/metamodel/util/HdfsResource.java @@ -0,0 +1,265 @@ +package org.apache.metamodel.util; + +import java.io.Closeable; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.metamodel.MetaModelException; + +/** + * A {@link Resource} implementation that connects to Apache Hadoop's HDFS + * distributed file system. + */ +public class HdfsResource implements Resource, Closeable { + + private static final Pattern URL_PATTERN = Pattern.compile("hdfs://(.+):([0-9]+)/(.*)"); + + private final String _hostname; + private final int _port; + private final String _filepath; + + private FileSystem _fileSystem; + + private Path _path; + + /** + * Creates a {@link HdfsResource} + * + * @param url + * a URL of the form: hdfs://hostname:port/path/to/file + */ + public HdfsResource(String url) { + if (url == null) { + throw new IllegalArgumentException("Url cannot be null"); + } + final Matcher matcher = URL_PATTERN.matcher(url); + if (!matcher.find()) { + throw new IllegalArgumentException("Cannot parse url '" + url + + "'. Must follow pattern: hdfs://hostname:port/path/to/file"); + } + _hostname = matcher.group(1); + _port = Integer.parseInt(matcher.group(2)); + _filepath = '/' + matcher.group(3); + } + + public HdfsResource(String hostname, int port, String filepath) { + _hostname = hostname; + _port = port; + _filepath = filepath; + } + + @Override + public String getName() { + final int lastSlash = _filepath.lastIndexOf('/'); + if (lastSlash != -1) { + return _filepath.substring(lastSlash + 1); + } + return _filepath; + } + + @Override + public String getQualifiedPath() { + return "hdfs://" + _hostname + ":" + _port + _filepath; + } + + @Override + public boolean isReadOnly() { + // TODO Auto-generated method stub + return false; + } + + @Override + public boolean isExists() { + return doWithFileSystem(new UncheckedFunc<FileSystem, Boolean>() { + @Override + protected Boolean evalUnchecked(FileSystem fs) throws Exception { + return fs.exists(getHadoopPath()); + } + }); + } + + @Override + public long getSize() { + return doWithFileSystem(new UncheckedFunc<FileSystem, Long>() { + @Override + protected Long evalUnchecked(FileSystem fs) throws Exception { + return fs.getFileStatus(getHadoopPath()).getLen(); + } + }); + } + + @Override + public long getLastModified() { + return doWithFileSystem(new UncheckedFunc<FileSystem, Long>() { + @Override + protected Long evalUnchecked(FileSystem fs) throws Exception { + return fs.getFileStatus(getHadoopPath()).getModificationTime(); + } + }); + } + + @Override + public void write(final Action<OutputStream> writeCallback) throws ResourceException { + final OutputStream out = doWithFileSystem(new UncheckedFunc<FileSystem, OutputStream>() { + @Override + protected OutputStream evalUnchecked(FileSystem fs) throws Exception { + return fs.create(getHadoopPath(), true); + } + }); + try { + writeCallback.run(out); + } catch (Exception e) { + throw wrapException(e); + } finally { + FileHelper.safeClose(out); + } + } + + @Override + public void append(Action<OutputStream> appendCallback) throws ResourceException { + final OutputStream out = doWithFileSystem(new UncheckedFunc<FileSystem, OutputStream>() { + @Override + protected OutputStream evalUnchecked(FileSystem fs) throws Exception { + return fs.append(getHadoopPath()); + } + }); + try { + appendCallback.run(out); + } catch (Exception e) { + throw wrapException(e); + } finally { + FileHelper.safeClose(out); + } + } + + @Override + public InputStream read() throws ResourceException { + return doWithFileSystem(new UncheckedFunc<FileSystem, InputStream>() { + @Override + protected InputStream evalUnchecked(FileSystem fs) throws Exception { + return fs.open(getHadoopPath()); + } + }); + } + + @Override + public void read(Action<InputStream> readCallback) throws ResourceException { + final InputStream in = read(); + try { + readCallback.run(in); + } catch (Exception e) { + throw wrapException(e); + } finally { + FileHelper.safeClose(in); + } + } + + @Override + public <E> E read(Func<InputStream, E> readCallback) throws ResourceException { + final InputStream in = read(); + try { + return readCallback.eval(in); + } catch (Exception e) { + throw wrapException(e); + } finally { + FileHelper.safeClose(in); + } + } + + private <E> E doWithFileSystem(Func<FileSystem, E> action) { + final FileSystem hadoopFileSystem = getHadoopFileSystem(); + try { + return action.eval(hadoopFileSystem); + } catch (Exception e) { + throw wrapException(e); + } + } + + private RuntimeException wrapException(Exception e) { + if (e instanceof RuntimeException) { + return (RuntimeException) e; + } + return new MetaModelException(e); + } + + public Configuration getHadoopConfiguration() { + final Configuration conf = new Configuration(); + conf.set("fs.defaultFS", "hdfs://" + _hostname + ":" + _port); + return conf; + } + + public FileSystem getHadoopFileSystem() { + if (_fileSystem == null) { + try { + _fileSystem = FileSystem.get(getHadoopConfiguration()); + } catch (IOException e) { + throw new MetaModelException("Could not connect to HDFS: " + e.getMessage(), e); + } + } + return _fileSystem; + } + + public Path getHadoopPath() { + if (_path == null) { + _path = new Path(_filepath); + } + return _path; + } + + @Override + public void close() throws IOException { + if (_fileSystem != null) { + try { + _fileSystem.close(); + } finally { + _fileSystem = null; + } + } + } + + @Override + protected void finalize() throws Throwable { + super.finalize(); + close(); + } + + @Override + public int hashCode() { + final int prime = 31; + int result = 1; + result = prime * result + ((_filepath == null) ? 0 : _filepath.hashCode()); + result = prime * result + ((_hostname == null) ? 0 : _hostname.hashCode()); + result = prime * result + _port; + return result; + } + + @Override + public boolean equals(Object obj) { + if (this == obj) + return true; + if (obj == null) + return false; + if (getClass() != obj.getClass()) + return false; + HdfsResource other = (HdfsResource) obj; + if (_filepath == null) { + if (other._filepath != null) + return false; + } else if (!_filepath.equals(other._filepath)) + return false; + if (_hostname == null) { + if (other._hostname != null) + return false; + } else if (!_hostname.equals(other._hostname)) + return false; + if (_port != other._port) + return false; + return true; + } +} http://git-wip-us.apache.org/repos/asf/metamodel/blob/9191b9fb/hadoop/src/test/java/org/apache/metamodel/util/HdfsResourceTest.java ---------------------------------------------------------------------- diff --git a/hadoop/src/test/java/org/apache/metamodel/util/HdfsResourceTest.java b/hadoop/src/test/java/org/apache/metamodel/util/HdfsResourceTest.java new file mode 100644 index 0000000..ebbbc39 --- /dev/null +++ b/hadoop/src/test/java/org/apache/metamodel/util/HdfsResourceTest.java @@ -0,0 +1,28 @@ +package org.apache.metamodel.util; + +import junit.framework.TestCase; + +public class HdfsResourceTest extends TestCase { + + public void testGetQualifiedName() throws Exception { + final HdfsResource res1 = new HdfsResource("hdfs://localhost:9000/home/metamodel.txt"); + assertEquals("hdfs://localhost:9000/home/metamodel.txt", res1.getQualifiedPath()); + assertEquals("metamodel.txt", res1.getName()); + + final HdfsResource res2 = new HdfsResource("localhost", 9000, "/home/metamodel.txt"); + assertEquals("hdfs://localhost:9000/home/metamodel.txt", res2.getQualifiedPath()); + assertEquals("metamodel.txt", res2.getName()); + + assertEquals(res1, res2); + + final HdfsResource res3 = new HdfsResource("localhost", 9000, "/home/apache.txt"); + assertEquals("hdfs://localhost:9000/home/apache.txt", res3.getQualifiedPath()); + assertEquals("apache.txt", res3.getName()); + + assertFalse(res3.equals(res1)); + + res1.close(); + res2.close(); + res3.close(); + } +} http://git-wip-us.apache.org/repos/asf/metamodel/blob/9191b9fb/hbase/src/main/java/org/apache/metamodel/hbase/HBaseDataContext.java ---------------------------------------------------------------------- diff --git a/hbase/src/main/java/org/apache/metamodel/hbase/HBaseDataContext.java b/hbase/src/main/java/org/apache/metamodel/hbase/HBaseDataContext.java index a501790..6cc1bc1 100644 --- a/hbase/src/main/java/org/apache/metamodel/hbase/HBaseDataContext.java +++ b/hbase/src/main/java/org/apache/metamodel/hbase/HBaseDataContext.java @@ -24,6 +24,7 @@ import java.util.List; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HTableDescriptor; +import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.Get; import org.apache.hadoop.hbase.client.HBaseAdmin; import org.apache.hadoop.hbase.client.HTableInterface;
