mirror of
https://github.com/apache/zeppelin
synced 2026-05-24 09:38:26 +00:00
Add File Interpreter, HDFS Interpreter and Tests
This commit is contained in:
Raj Bains
parent
26c89a3364
commit
865e6ab597
6 changed files with 914 additions and 0 deletions
157
file/pom.xml
Normal file
157
file/pom.xml
Normal file
|
|
@ -0,0 +1,157 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!--
|
||||
~ 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.
|
||||
-->
|
||||
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<artifactId>zeppelin</artifactId>
|
||||
<groupId>org.apache.zeppelin</groupId>
|
||||
<version>0.6.0-incubating-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<groupId>org.apache.zeppelin</groupId>
|
||||
<artifactId>zeppelin-file</artifactId>
|
||||
<packaging>jar</packaging>
|
||||
<version>0.6.0-incubating-SNAPSHOT</version>
|
||||
<name>Zeppelin: File Manager</name>
|
||||
<url>http://www.apache.org</url>
|
||||
|
||||
<properties>
|
||||
<hive.hive.version>0.14.0</hive.hive.version>
|
||||
<hive.hadoop.version>2.6.0</hive.hadoop.version>
|
||||
</properties>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.apache.zeppelin</groupId>
|
||||
<artifactId>zeppelin-interpreter</artifactId>
|
||||
<version>${project.version}</version>
|
||||
<scope>provided</scope>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.apache.commons</groupId>
|
||||
<artifactId>commons-exec</artifactId>
|
||||
<version>1.1</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.slf4j</groupId>
|
||||
<artifactId>slf4j-api</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.slf4j</groupId>
|
||||
<artifactId>slf4j-log4j12</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.apache.hadoop</groupId>
|
||||
<artifactId>hadoop-common</artifactId>
|
||||
<version>${hive.hadoop.version}</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.apache.hadoop</groupId>
|
||||
<artifactId>hadoop-hdfs</artifactId>
|
||||
<version>${hive.hadoop.version}</version>
|
||||
</dependency>
|
||||
|
||||
|
||||
|
||||
<dependency>
|
||||
<groupId>junit</groupId>
|
||||
<artifactId>junit</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-deploy-plugin</artifactId>
|
||||
<version>2.7</version>
|
||||
<configuration>
|
||||
<skip>true</skip>
|
||||
</configuration>
|
||||
</plugin>
|
||||
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-surefire-plugin</artifactId>
|
||||
<version>2.18.1</version>
|
||||
</plugin>
|
||||
|
||||
<plugin>
|
||||
<artifactId>maven-enforcer-plugin</artifactId>
|
||||
<version>1.3.1</version>
|
||||
<executions>
|
||||
<execution>
|
||||
<id>enforce</id>
|
||||
<phase>none</phase>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
|
||||
<plugin>
|
||||
<artifactId>maven-dependency-plugin</artifactId>
|
||||
<version>2.8</version>
|
||||
<executions>
|
||||
<execution>
|
||||
<id>copy-dependencies</id>
|
||||
<phase>package</phase>
|
||||
<goals>
|
||||
<goal>copy-dependencies</goal>
|
||||
</goals>
|
||||
<configuration>
|
||||
<outputDirectory>${project.build.directory}/../../interpreter/file</outputDirectory>
|
||||
<overWriteReleases>false</overWriteReleases>
|
||||
<overWriteSnapshots>false</overWriteSnapshots>
|
||||
<overWriteIfNewer>true</overWriteIfNewer>
|
||||
<includeScope>runtime</includeScope>
|
||||
</configuration>
|
||||
</execution>
|
||||
<execution>
|
||||
<id>copy-artifact</id>
|
||||
<phase>package</phase>
|
||||
<goals>
|
||||
<goal>copy</goal>
|
||||
</goals>
|
||||
<configuration>
|
||||
<outputDirectory>${project.build.directory}/../../interpreter/file</outputDirectory>
|
||||
<overWriteReleases>false</overWriteReleases>
|
||||
<overWriteSnapshots>false</overWriteSnapshots>
|
||||
<overWriteIfNewer>true</overWriteIfNewer>
|
||||
<!--<includeScope>runtime</includeScope>-->
|
||||
<artifactItems>
|
||||
<artifactItem>
|
||||
<groupId>${project.groupId}</groupId>
|
||||
<artifactId>${project.artifactId}</artifactId>
|
||||
<version>${project.version}</version>
|
||||
<type>${project.packaging}</type>
|
||||
</artifactItem>
|
||||
</artifactItems>
|
||||
</configuration>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
</project>
|
||||
170
file/src/main/java/org/apache/zeppelin/file/FileInterpreter.java
Normal file
170
file/src/main/java/org/apache/zeppelin/file/FileInterpreter.java
Normal file
|
|
@ -0,0 +1,170 @@
|
|||
/**
|
||||
* 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.zeppelin.file;
|
||||
|
||||
import org.apache.zeppelin.interpreter.Interpreter;
|
||||
import org.apache.zeppelin.interpreter.InterpreterContext;
|
||||
import org.apache.zeppelin.interpreter.InterpreterResult;
|
||||
import org.apache.zeppelin.interpreter.InterpreterResult.Code;
|
||||
import org.apache.zeppelin.interpreter.InterpreterResult.Type;
|
||||
import org.apache.zeppelin.scheduler.Scheduler;
|
||||
import org.apache.zeppelin.scheduler.SchedulerFactory;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.Paths;
|
||||
import java.util.*;
|
||||
|
||||
/**
|
||||
* File interpreter for Zeppelin.
|
||||
*
|
||||
* @author rajbains
|
||||
*/
|
||||
public abstract class FileInterpreter extends Interpreter {
|
||||
Logger logger = LoggerFactory.getLogger(FileInterpreter.class);
|
||||
String currentDir = null;
|
||||
CommandArgs args = null;
|
||||
|
||||
public FileInterpreter(Properties property) {
|
||||
super(property);
|
||||
currentDir = new String("/");
|
||||
}
|
||||
|
||||
/**
|
||||
* Handling the arguments of the command
|
||||
*/
|
||||
public class CommandArgs {
|
||||
public String input = null;
|
||||
public String command = null;
|
||||
public ArrayList<String> args = null;
|
||||
public HashSet<Character> flags = null;
|
||||
|
||||
public CommandArgs(String cmd) {
|
||||
input = cmd;
|
||||
args = new ArrayList();
|
||||
flags = new HashSet();
|
||||
}
|
||||
|
||||
private void parseArg(String arg) {
|
||||
if (arg.charAt(0) == '-') { // handle flags
|
||||
for (int i = 0; i < arg.length(); i++) {
|
||||
Character c = arg.charAt(i);
|
||||
flags.add(c);
|
||||
}
|
||||
} else { // handle other args
|
||||
args.add(arg);
|
||||
}
|
||||
}
|
||||
|
||||
public void parseArgs() {
|
||||
if (input == null)
|
||||
return;
|
||||
StringTokenizer st = new StringTokenizer(input);
|
||||
if (st.hasMoreTokens()) {
|
||||
command = st.nextToken();
|
||||
while (st.hasMoreTokens())
|
||||
parseArg(st.nextToken());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Functions that each file system implementation must override
|
||||
|
||||
public abstract String listAll(String path);
|
||||
|
||||
public abstract boolean isDirectory(String path);
|
||||
|
||||
// Combine paths, takes care of arguments such as ..
|
||||
|
||||
private String getNewPath(String argument){
|
||||
Path arg = Paths.get(argument);
|
||||
Path ret = arg.isAbsolute() ? arg : Paths.get(currentDir, argument);
|
||||
return ret.normalize().toString();
|
||||
}
|
||||
|
||||
// Handle the command handling uniformly across all file systems
|
||||
|
||||
@Override
|
||||
public InterpreterResult interpret(String cmd, InterpreterContext contextInterpreter) {
|
||||
logger.info("Run File command '" + cmd + "'");
|
||||
|
||||
args = new CommandArgs(cmd);
|
||||
args.parseArgs();
|
||||
|
||||
if (args.command == null) {
|
||||
logger.info("Error: No command");
|
||||
return new InterpreterResult(Code.ERROR, Type.TEXT, "No command");
|
||||
}
|
||||
|
||||
// Simple parsing of the command
|
||||
|
||||
if (args.command.equals("cd")) {
|
||||
|
||||
String newPath = !args.args.isEmpty() ? getNewPath(args.args.get(0)) : currentDir;
|
||||
if (!isDirectory(newPath))
|
||||
return new InterpreterResult(Code.ERROR, Type.TEXT, "Invalid Directory");
|
||||
|
||||
currentDir = newPath;
|
||||
return new InterpreterResult(Code.SUCCESS, Type.TEXT, "OK");
|
||||
|
||||
} else if (args.command.equals("ls")) {
|
||||
|
||||
String newPath = !args.args.isEmpty() ? getNewPath(args.args.get(0)) : currentDir;
|
||||
if (!isDirectory(newPath))
|
||||
return new InterpreterResult(Code.ERROR, Type.TEXT, "Invalid List Directory");
|
||||
|
||||
String results = listAll(newPath);
|
||||
return new InterpreterResult(Code.SUCCESS, Type.TEXT, results);
|
||||
|
||||
} else if (args.command.equals("pwd")) {
|
||||
|
||||
return new InterpreterResult(Code.SUCCESS, Type.TEXT, currentDir);
|
||||
|
||||
} else {
|
||||
|
||||
return new InterpreterResult(Code.ERROR, Type.TEXT, "Unknown command");
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cancel(InterpreterContext context) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public FormType getFormType() {
|
||||
return FormType.SIMPLE;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getProgress(InterpreterContext context) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Scheduler getScheduler() {
|
||||
return SchedulerFactory.singleton().createOrGetFIFOScheduler(
|
||||
FileInterpreter.class.getName() + this.hashCode());
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<String> completion(String buf, int cursor) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
151
file/src/main/java/org/apache/zeppelin/file/HDFSCommand.java
Normal file
151
file/src/main/java/org/apache/zeppelin/file/HDFSCommand.java
Normal file
|
|
@ -0,0 +1,151 @@
|
|||
/**
|
||||
* 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.zeppelin.file;
|
||||
|
||||
import java.net.URL;
|
||||
import java.net.HttpURLConnection;
|
||||
import java.io.BufferedReader;
|
||||
import java.io.InputStreamReader;
|
||||
import javax.ws.rs.core.UriBuilder;
|
||||
import org.slf4j.Logger;
|
||||
|
||||
/**
|
||||
* Definition and HTTP invocation methods for all WebHDFS commands
|
||||
*
|
||||
* @author rajbains
|
||||
*
|
||||
*/
|
||||
public class HDFSCommand {
|
||||
|
||||
/**
|
||||
* Type of HTTP request
|
||||
*/
|
||||
public enum HttpType {
|
||||
GET,
|
||||
PUT
|
||||
}
|
||||
|
||||
/**
|
||||
* Definition of WebHDFS operator
|
||||
*/
|
||||
public class Op {
|
||||
public String op;
|
||||
public HttpType cmd;
|
||||
public int minArgs;
|
||||
|
||||
public Op(String op, HttpType cmd, int minArgs) {
|
||||
this.op = op;
|
||||
this.cmd = cmd;
|
||||
this.minArgs = minArgs;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Definition of argument to an operator
|
||||
*/
|
||||
public class Arg {
|
||||
public String key;
|
||||
public String value;
|
||||
|
||||
public Arg(String key, String value) {
|
||||
this.key = key;
|
||||
this.value = value;
|
||||
}
|
||||
}
|
||||
|
||||
// How to connect to WebHDFS
|
||||
String url = null;
|
||||
String user = null;
|
||||
Logger logger;
|
||||
|
||||
// Define all the commands available
|
||||
public Op getFileStatus = new Op("GETFILESTATUS", HttpType.GET, 0);
|
||||
public Op listStatus = new Op("LISTSTATUS", HttpType.GET, 0);
|
||||
|
||||
public HDFSCommand(String url, String user, Logger logger) {
|
||||
super();
|
||||
this.url = url;
|
||||
this.user = user;
|
||||
this.logger = logger;
|
||||
}
|
||||
|
||||
public String checkArgs(Op op, String path, Arg[] args) throws Exception {
|
||||
if (op == null ||
|
||||
path == null ||
|
||||
(op.minArgs > 0 &&
|
||||
(args == null ||
|
||||
args.length != op.minArgs)))
|
||||
{
|
||||
String a = "";
|
||||
a = (op != null) ? a + op.op + "\n" : a;
|
||||
a = (path != null) ? a + path + "\n" : a;
|
||||
a = (args != null) ? a + args + "\n" : a;
|
||||
return a;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
|
||||
// The operator that runs all commands
|
||||
public String runCommand(Op op, String path, Arg[] args) throws Exception {
|
||||
|
||||
// Check arguments
|
||||
String error = checkArgs(op, path, args);
|
||||
if (error != null) {
|
||||
logger.error("Bad arguments to command: " + error);
|
||||
return "ERROR: BAD ARGS";
|
||||
}
|
||||
|
||||
// Build URI
|
||||
UriBuilder builder = UriBuilder
|
||||
.fromPath(url)
|
||||
.path(path)
|
||||
.queryParam("op", op.op);
|
||||
|
||||
if (args != null) {
|
||||
for (Arg a : args) {
|
||||
builder = builder.queryParam(a.key, a.value);
|
||||
}
|
||||
}
|
||||
java.net.URI uri = builder.build();
|
||||
|
||||
// Connect and get response string
|
||||
URL hdfsUrl = uri.toURL();
|
||||
HttpURLConnection con = (HttpURLConnection) hdfsUrl.openConnection();
|
||||
|
||||
if (op.cmd == HttpType.GET) {
|
||||
con.setRequestMethod("GET");
|
||||
int responseCode = con.getResponseCode();
|
||||
logger.info("Sending 'GET' request to URL : " + hdfsUrl);
|
||||
logger.info("Response Code : " + responseCode);
|
||||
|
||||
BufferedReader in = new BufferedReader(
|
||||
new InputStreamReader(con.getInputStream()));
|
||||
String inputLine;
|
||||
StringBuffer response = new StringBuffer();
|
||||
|
||||
while ((inputLine = in.readLine()) != null) {
|
||||
response.append(inputLine);
|
||||
}
|
||||
in.close();
|
||||
return response.toString();
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,225 @@
|
|||
/**
|
||||
* 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.zeppelin.file;
|
||||
|
||||
import java.text.SimpleDateFormat;
|
||||
import java.util.Date;
|
||||
import java.util.Properties;
|
||||
import com.google.gson.Gson;
|
||||
import org.apache.zeppelin.interpreter.Interpreter;
|
||||
import org.apache.zeppelin.interpreter.InterpreterPropertyBuilder;
|
||||
|
||||
/**
|
||||
* HDFS implementation of File interpreter for Zeppelin.
|
||||
*
|
||||
* @author rajbains
|
||||
*
|
||||
*/
|
||||
public class HDFSFileInterpreter extends FileInterpreter {
|
||||
static final String HDFS_URL = "hdfs.url";
|
||||
static final String HDFS_USER = "hdfs.user";
|
||||
|
||||
static {
|
||||
Interpreter.register(
|
||||
"hdfs",
|
||||
"hdfs",
|
||||
HDFSFileInterpreter.class.getName(),
|
||||
new InterpreterPropertyBuilder()
|
||||
.add(HDFS_URL, "http://c6401.ambari.apache.org:50070/webhdfs/v1/",
|
||||
"The URL for WebHDFS")
|
||||
.add(HDFS_USER, "hdfs", "The WebHDFS user").build());
|
||||
}
|
||||
|
||||
Exception exceptionOnConnect = null;
|
||||
HDFSCommand cmd = null;
|
||||
Gson gson = null;
|
||||
|
||||
public void prepare() {
|
||||
String userName = getProperty(HDFS_USER);
|
||||
String hdfsUrl = getProperty(HDFS_URL);
|
||||
cmd = new HDFSCommand(hdfsUrl, userName, logger);
|
||||
gson = new Gson();
|
||||
}
|
||||
|
||||
public HDFSFileInterpreter(Properties property){
|
||||
super(property);
|
||||
prepare();
|
||||
}
|
||||
|
||||
/**
|
||||
* Status of one file
|
||||
*
|
||||
* matches returned JSON
|
||||
*/
|
||||
public class OneFileStatus {
|
||||
public long accessTime;
|
||||
public int blockSize;
|
||||
public int childrenNum;
|
||||
public int fileId;
|
||||
public String group;
|
||||
public long length;
|
||||
public long modificationTime;
|
||||
public String owner;
|
||||
public String pathSuffix;
|
||||
public String permission;
|
||||
public int replication;
|
||||
public int storagePolicy;
|
||||
public String type;
|
||||
public String toString() {
|
||||
String str = "";
|
||||
str += "\nAccessTime = " + accessTime;
|
||||
str += "\nBlockSize = " + blockSize;
|
||||
str += "\nChildrenNum = " + childrenNum;
|
||||
str += "\nFileId = " + fileId;
|
||||
str += "\nGroup = " + group;
|
||||
str += "\nLength = " + length;
|
||||
str += "\nModificationTime = " + modificationTime;
|
||||
str += "\nOwner = " + owner;
|
||||
str += "\nPathSuffix = " + pathSuffix;
|
||||
str += "\nPermission = " + permission;
|
||||
str += "\nReplication = " + replication;
|
||||
str += "\nStoragePolicy = " + storagePolicy;
|
||||
str += "\nType = " + type;
|
||||
return str;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Status of one file
|
||||
*
|
||||
* matches returned JSON
|
||||
*/
|
||||
public class SingleFileStatus {
|
||||
public OneFileStatus FileStatus;
|
||||
}
|
||||
|
||||
/**
|
||||
* Status of all files in a directory
|
||||
*
|
||||
* matches returned JSON
|
||||
*/
|
||||
public class MultiFileStatus {
|
||||
public OneFileStatus[] FileStatus;
|
||||
}
|
||||
|
||||
/**
|
||||
* Status of all files in a directory
|
||||
*
|
||||
* matches returned JSON
|
||||
*/
|
||||
public class AllFileStatus {
|
||||
public MultiFileStatus FileStatuses;
|
||||
}
|
||||
|
||||
// tests whether we're able to connect to HDFS
|
||||
|
||||
private void testConnection() {
|
||||
try {
|
||||
if (isDirectory("/"))
|
||||
logger.info("Successfully created WebHDFS connection");
|
||||
} catch (Exception e) {
|
||||
logger.error("testConnection: Cannot open WebHDFS connection. Bad URL: " + "/", e);
|
||||
exceptionOnConnect = e;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void open() {
|
||||
testConnection();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
}
|
||||
|
||||
private String listDir(String path) throws Exception {
|
||||
return cmd.runCommand(cmd.listStatus, path, null);
|
||||
}
|
||||
|
||||
private String listPermission(OneFileStatus fs){
|
||||
String s = "";
|
||||
s += fs.type.equalsIgnoreCase("Directory") ? 'd' : '-';
|
||||
int p = Integer.parseInt(fs.permission, 16);
|
||||
s += ((p & 0x400) == 0) ? '-' : 'r';
|
||||
s += ((p & 0x200) == 0) ? '-' : 'w';
|
||||
s += ((p & 0x100) == 0) ? '-' : 'x';
|
||||
s += ((p & 0x40) == 0) ? '-' : 'r';
|
||||
s += ((p & 0x20) == 0) ? '-' : 'w';
|
||||
s += ((p & 0x10) == 0) ? '-' : 'x';
|
||||
s += ((p & 0x4) == 0) ? '-' : 'r';
|
||||
s += ((p & 0x2) == 0) ? '-' : 'w';
|
||||
s += ((p & 0x1) == 0) ? '-' : 'x';
|
||||
return s;
|
||||
}
|
||||
private String listDate(OneFileStatus fs) {
|
||||
return new SimpleDateFormat("yyyy-MM-dd HH:mm").format(new Date(fs.modificationTime));
|
||||
}
|
||||
private String ListOne(String path, OneFileStatus fs) {
|
||||
if (args.flags.contains(new Character('l'))) {
|
||||
String s = "";
|
||||
s += listPermission(fs) + "\t ";
|
||||
s += ((fs.replication == 0) ? "-" : fs.replication) + "\t ";
|
||||
s += fs.owner + "\t ";
|
||||
s += fs.group + "\t ";
|
||||
s += fs.length + "\t ";
|
||||
s += listDate(fs) + " GMT\t ";
|
||||
s += (path.length() == 1) ? path + fs.pathSuffix : path + '/' + fs.pathSuffix;
|
||||
return s;
|
||||
}
|
||||
return fs.pathSuffix;
|
||||
}
|
||||
|
||||
public String listAll(String path) {
|
||||
String all = "";
|
||||
if (exceptionOnConnect != null)
|
||||
return all;
|
||||
try {
|
||||
String sfs = listDir(path);
|
||||
if (sfs != null) {
|
||||
AllFileStatus allFiles = gson.fromJson(sfs, AllFileStatus.class);
|
||||
|
||||
if (allFiles != null &&
|
||||
allFiles.FileStatuses != null &&
|
||||
allFiles.FileStatuses.FileStatus != null)
|
||||
{
|
||||
for (OneFileStatus fs : allFiles.FileStatuses.FileStatus)
|
||||
all = all + ListOne(path, fs) + '\n';
|
||||
}
|
||||
}
|
||||
} catch (Exception e) {
|
||||
logger.error("listall: listDir " + path, e);
|
||||
}
|
||||
return all;
|
||||
}
|
||||
|
||||
public boolean isDirectory(String path) {
|
||||
boolean ret = false;
|
||||
if (exceptionOnConnect != null)
|
||||
return ret;
|
||||
try {
|
||||
String str = cmd.runCommand(cmd.getFileStatus, path, null);
|
||||
SingleFileStatus sfs = gson.fromJson(str, SingleFileStatus.class);
|
||||
if (sfs != null)
|
||||
return sfs.FileStatus.type.equals("DIRECTORY");
|
||||
} catch (Exception e) {
|
||||
logger.error("IsDirectory: " + path, e);
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,210 @@
|
|||
/**
|
||||
* 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.zeppelin.file;
|
||||
|
||||
import com.google.gson.Gson;
|
||||
import junit.framework.TestCase;
|
||||
import static org.junit.Assert.*;
|
||||
import org.apache.zeppelin.interpreter.InterpreterResult;
|
||||
import org.junit.Test;
|
||||
import org.slf4j.Logger;
|
||||
import java.util.HashMap;
|
||||
import java.util.Properties;
|
||||
import java.lang.Override;
|
||||
import java.lang.String;
|
||||
|
||||
|
||||
/**
|
||||
* Tests Hive Interpreter by running pre-determined commands against mock file system
|
||||
*
|
||||
* Created by rajbains on 8/29/15.
|
||||
*/
|
||||
public class HDFSFileInterpreterTest extends TestCase {
|
||||
|
||||
@Test
|
||||
public void test() {
|
||||
HDFSFileInterpreter t = new MockHDFSFileInterpreter(new Properties());
|
||||
t.open();
|
||||
|
||||
// We have info for /, /user, /tmp, /mr-history/done
|
||||
|
||||
// Ensure
|
||||
// 1. ls -l works
|
||||
// 2. paths (. and ..) are correctly handled
|
||||
// 3. flags and arguments to commands are correctly handled
|
||||
|
||||
InterpreterResult result1 = t.interpret("ls -l /", null);
|
||||
assertEquals(result1.type(), InterpreterResult.Type.TEXT);
|
||||
|
||||
InterpreterResult result2 = t.interpret("ls -l /./user/..", null);
|
||||
assertEquals(result2.type(), InterpreterResult.Type.TEXT);
|
||||
|
||||
assertEquals(result1.message(), result2.message());
|
||||
|
||||
// Ensure you can do cd and after that the ls uses current directory correctly
|
||||
|
||||
InterpreterResult result3 = t.interpret("cd user", null);
|
||||
assertEquals(result3.type(), InterpreterResult.Type.TEXT);
|
||||
assertEquals(result3.message(), "OK");
|
||||
|
||||
InterpreterResult result4 = t.interpret("ls", null);
|
||||
assertEquals(result4.type(), InterpreterResult.Type.TEXT);
|
||||
|
||||
InterpreterResult result5 = t.interpret("ls /user", null);
|
||||
assertEquals(result5.type(), InterpreterResult.Type.TEXT);
|
||||
|
||||
assertEquals(result4.message(), result5.message());
|
||||
|
||||
// Ensure pwd works correctly
|
||||
|
||||
InterpreterResult result6 = t.interpret("pwd", null);
|
||||
assertEquals(result6.type(), InterpreterResult.Type.TEXT);
|
||||
assertEquals(result6.message(), "/user");
|
||||
|
||||
// Move a couple of levels and check we're in the right place
|
||||
|
||||
InterpreterResult result7 = t.interpret("cd ../mr-history/done", null);
|
||||
assertEquals(result7.type(), InterpreterResult.Type.TEXT);
|
||||
assertEquals(result7.message(), "OK");
|
||||
|
||||
InterpreterResult result8 = t.interpret("ls -l ", null);
|
||||
assertEquals(result8.type(), InterpreterResult.Type.TEXT);
|
||||
|
||||
InterpreterResult result9 = t.interpret("ls -l /mr-history/done", null);
|
||||
assertEquals(result9.type(), InterpreterResult.Type.TEXT);
|
||||
|
||||
assertEquals(result8.message(), result9.message());
|
||||
|
||||
InterpreterResult result10 = t.interpret("cd ../..", null);
|
||||
assertEquals(result10.type(), InterpreterResult.Type.TEXT);
|
||||
assertEquals(result7.message(), "OK");
|
||||
|
||||
InterpreterResult result11 = t.interpret("ls -l ", null);
|
||||
assertEquals(result11.type(), InterpreterResult.Type.TEXT);
|
||||
|
||||
// we should be back to first result after all this navigation
|
||||
assertEquals(result1.message(), result11.message());
|
||||
|
||||
t.close();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Store command results from curl against a real file system
|
||||
*/
|
||||
class MockFileSystem {
|
||||
HashMap<String, String> mfs = new HashMap<String, String>();
|
||||
void addListStatusData() {
|
||||
mfs.put("/?op=LISTSTATUS",
|
||||
"{\"FileStatuses\":{\"FileStatus\":[\n" +
|
||||
"{\"accessTime\":0,\"blockSize\":0,\"childrenNum\":1,\"fileId\":16389,\"group\":\"hadoop\",\"length\":0,\"modificationTime\":1438548219672,\"owner\":\"yarn\",\"pathSuffix\":\"app-logs\",\"permission\":\"777\",\"replication\":0,\"storagePolicy\":0,\"type\":\"DIRECTORY\"},\n" +
|
||||
"{\"accessTime\":0,\"blockSize\":0,\"childrenNum\":1,\"fileId\":16395,\"group\":\"hdfs\",\"length\":0,\"modificationTime\":1438548030045,\"owner\":\"hdfs\",\"pathSuffix\":\"hdp\",\"permission\":\"755\",\"replication\":0,\"storagePolicy\":0,\"type\":\"DIRECTORY\"},\n" +
|
||||
"{\"accessTime\":0,\"blockSize\":0,\"childrenNum\":1,\"fileId\":16390,\"group\":\"hdfs\",\"length\":0,\"modificationTime\":1438547985336,\"owner\":\"mapred\",\"pathSuffix\":\"mapred\",\"permission\":\"755\",\"replication\":0,\"storagePolicy\":0,\"type\":\"DIRECTORY\"},\n" +
|
||||
"{\"accessTime\":0,\"blockSize\":0,\"childrenNum\":2,\"fileId\":16392,\"group\":\"hdfs\",\"length\":0,\"modificationTime\":1438547985346,\"owner\":\"hdfs\",\"pathSuffix\":\"mr-history\",\"permission\":\"755\",\"replication\":0,\"storagePolicy\":0,\"type\":\"DIRECTORY\"},\n" +
|
||||
"{\"accessTime\":0,\"blockSize\":0,\"childrenNum\":1,\"fileId\":16400,\"group\":\"hdfs\",\"length\":0,\"modificationTime\":1438548089725,\"owner\":\"hdfs\",\"pathSuffix\":\"system\",\"permission\":\"755\",\"replication\":0,\"storagePolicy\":0,\"type\":\"DIRECTORY\"},\n" +
|
||||
"{\"accessTime\":0,\"blockSize\":0,\"childrenNum\":1,\"fileId\":16386,\"group\":\"hdfs\",\"length\":0,\"modificationTime\":1438548150089,\"owner\":\"hdfs\",\"pathSuffix\":\"tmp\",\"permission\":\"777\",\"replication\":0,\"storagePolicy\":0,\"type\":\"DIRECTORY\"},\n" +
|
||||
"{\"accessTime\":0,\"blockSize\":0,\"childrenNum\":1,\"fileId\":16387,\"group\":\"hdfs\",\"length\":0,\"modificationTime\":1438547921792,\"owner\":\"hdfs\",\"pathSuffix\":\"user\",\"permission\":\"755\",\"replication\":0,\"storagePolicy\":0,\"type\":\"DIRECTORY\"}\n" +
|
||||
"]}}"
|
||||
);
|
||||
mfs.put("/user?op=LISTSTATUS",
|
||||
"{\"FileStatuses\":{\"FileStatus\":[\n" +
|
||||
" {\"accessTime\":0,\"blockSize\":0,\"childrenNum\":4,\"fileId\":16388,\"group\":\"hdfs\",\"length\":0,\"modificationTime\":1441253161263,\"owner\":\"ambari-qa\",\"pathSuffix\":\"ambari-qa\",\"permission\":\"770\",\"replication\":0,\"storagePolicy\":0,\"type\":\"DIRECTORY\"}\n" +
|
||||
" ]}}"
|
||||
);
|
||||
mfs.put("/tmp?op=LISTSTATUS",
|
||||
"{\"FileStatuses\":{\"FileStatus\":[\n" +
|
||||
" {\"accessTime\":1441253097489,\"blockSize\":134217728,\"childrenNum\":0,\"fileId\":16400,\"group\":\"hdfs\",\"length\":1645,\"modificationTime\":1441253097517,\"owner\":\"hdfs\",\"pathSuffix\":\"ida8c06540_date040315\",\"permission\":\"755\",\"replication\":3,\"storagePolicy\":0,\"type\":\"FILE\"}\n" +
|
||||
" ]}}"
|
||||
);
|
||||
mfs.put("/mr-history/done?op=LISTSTATUS",
|
||||
"{\"FileStatuses\":{\"FileStatus\":[\n" +
|
||||
"{\"accessTime\":0,\"blockSize\":0,\"childrenNum\":1,\"fileId\":16433,\"group\":\"hadoop\",\"length\":0,\"modificationTime\":1441253197481,\"owner\":\"mapred\",\"pathSuffix\":\"2015\",\"permission\":\"770\",\"replication\":0,\"storagePolicy\":0,\"type\":\"DIRECTORY\"}\n" +
|
||||
"]}}"
|
||||
);
|
||||
}
|
||||
void addGetFileStatusData() {
|
||||
mfs.put("/?op=GETFILESTATUS",
|
||||
"{\"FileStatus\":{\"accessTime\":0,\"blockSize\":0,\"childrenNum\":7,\"fileId\":16385,\"group\":\"hdfs\",\"length\":0,\"modificationTime\":1438548089725,\"owner\":\"hdfs\",\"pathSuffix\":\"\",\"permission\":\"755\",\"replication\":0,\"storagePolicy\":0,\"type\":\"DIRECTORY\"}}");
|
||||
mfs.put("/user?op=GETFILESTATUS",
|
||||
"{\"FileStatus\":{\"accessTime\":0,\"blockSize\":0,\"childrenNum\":1,\"fileId\":16387,\"group\":\"hdfs\",\"length\":0,\"modificationTime\":1441253043188,\"owner\":\"hdfs\",\"pathSuffix\":\"\",\"permission\":\"755\",\"replication\":0,\"storagePolicy\":0,\"type\":\"DIRECTORY\"}}");
|
||||
mfs.put("/tmp?op=GETFILESTATUS",
|
||||
"{\"FileStatus\":{\"accessTime\":0,\"blockSize\":0,\"childrenNum\":1,\"fileId\":16386,\"group\":\"hdfs\",\"length\":0,\"modificationTime\":1441253097489,\"owner\":\"hdfs\",\"pathSuffix\":\"\",\"permission\":\"777\",\"replication\":0,\"storagePolicy\":0,\"type\":\"DIRECTORY\"}}");
|
||||
mfs.put("/mr-history/done?op=GETFILESTATUS",
|
||||
"{\"FileStatus\":{\"accessTime\":0,\"blockSize\":0,\"childrenNum\":1,\"fileId\":16393,\"group\":\"hadoop\",\"length\":0,\"modificationTime\":1441253197480,\"owner\":\"mapred\",\"pathSuffix\":\"\",\"permission\":\"777\",\"replication\":0,\"storagePolicy\":0,\"type\":\"DIRECTORY\"}}");
|
||||
}
|
||||
public void addMockData(HDFSCommand.Op op) {
|
||||
if (op.op.equals("LISTSTATUS")) {
|
||||
addListStatusData();
|
||||
} else if (op.op.equals("GETFILESTATUS")) {
|
||||
addGetFileStatusData();
|
||||
}
|
||||
// do nothing
|
||||
}
|
||||
public String get(String key) {
|
||||
return mfs.get(key);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Run commands against mock file system that simulates webhdfs responses
|
||||
*/
|
||||
class MockHDFSCommand extends HDFSCommand {
|
||||
MockFileSystem fs = null;
|
||||
|
||||
public MockHDFSCommand(String url, String user, Logger logger) {
|
||||
super(url, user, logger);
|
||||
fs = new MockFileSystem();
|
||||
fs.addMockData(getFileStatus);
|
||||
fs.addMockData(listStatus);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String runCommand(Op op, String path, Arg[] args) throws Exception {
|
||||
|
||||
String error = checkArgs(op, path, args);
|
||||
assertNull(error);
|
||||
|
||||
String c = path + "?op=" + op.op;
|
||||
|
||||
if (args != null) {
|
||||
for (Arg a : args) {
|
||||
c += "&" + a.key + "=" + a.value;
|
||||
}
|
||||
}
|
||||
return fs.get(c);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Mock Interpreter - uses Mock HDFS command
|
||||
*/
|
||||
class MockHDFSFileInterpreter extends HDFSFileInterpreter {
|
||||
|
||||
@Override
|
||||
public void prepare() {
|
||||
// Run commands against mock File System instead of WebHDFS
|
||||
cmd = new MockHDFSCommand("", "", logger);
|
||||
gson = new Gson();
|
||||
}
|
||||
|
||||
public MockHDFSFileInterpreter(Properties property) {
|
||||
super(property);
|
||||
}
|
||||
|
||||
}
|
||||
1
pom.xml
1
pom.xml
|
|
@ -96,6 +96,7 @@
|
|||
<module>geode</module>
|
||||
<module>postgresql</module>
|
||||
<module>tajo</module>
|
||||
<module>file</module>
|
||||
<module>flink</module>
|
||||
<module>ignite</module>
|
||||
<module>kylin</module>
|
||||
|
|
|
|||
Loading…
Reference in a new issue