[CARBONDATA-2752][CARBONSTORE] Carbon provide Zeppelin support

This closes #2522
apache · Jul 23, 2018 · 7ad2fd9 · 7ad2fd9
1 parent 9ac55a5
commit 7ad2fd9
Show file tree

Hide file tree

Showing 11 changed files with 755 additions and 12 deletions.
diff --git a/integration/zeppelin/README.md b/integration/zeppelin/README.md
@@ -0,0 +1,44 @@
+<!--
+    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.
+-->
+
+### Please follow below steps to integrate with zeppelin  
+1. run ```mvn package -Pzeppelin```
+	This will generate _carbondata-zeppelin-*.tar.gz_ under target folder
+2. Extract the tar content to _ZEPPELIN_INSTALL_HOME/interpreter/_
+3. Add _org.apache.carbonndata.zeppelin.CarbonInterpreter_ to list of interpreters mentioned by _zeppelin.interpreters_ @ _ZEPPELIN_INSTALL_HOME/conf/zeppelin-site.xml_ (create if not exists)
+	Example:
+```xml
+	<property>
+	  <name>zeppelin.interpreters</name>
+<value>org.apache.zeppelin.spark.SparkInterpreter,.....,org.apache.carbonndata.zeppelin.CarbonInterpreter</value>
+	  <description>Comma separated interpreter configurations. First interpreter become a default</description>
+	</property>
+```
+4. Add carbon to list of interpreters mentioned by zeppelin.interpreter.order @ ZEPPELIN_INSTALL_HOME/conf/zeppelin-site.xml
+	Example:
+```xml
+	<property>
+	  <name>zeppelin.interpreter.group.order</name>
+	  <value>spark,..,carbon</value>
+	  <description></description>
+	</property>
+```
+
+5. Restart Zeppelin server and add new interpreter with name _carbon_ from zeppelin interpreter page
+    Refer : https://zeppelin.apache.org/docs/0.8.0/usage/interpreter/overview.html#what-is-zeppelin-interpreter
+6. Configure ```carbon.query.api.url``` in interpreter setting from zeppelin interpreter page and click save
+7. Now can use notebook with interpreter ```%carbon```
diff --git a/integration/zeppelin/assembly/assembly.xml b/integration/zeppelin/assembly/assembly.xml
@@ -0,0 +1,37 @@
+<assembly>
+<id>compress</id>
+  <formats>
+    <format>tar.gz</format>
+  </formats>
+
+  <includeBaseDirectory>false</includeBaseDirectory>
+  <dependencySets>
+    <dependencySet>
+      <outputDirectory>/carbon</outputDirectory>
+	  <useTransitiveDependencies>false</useTransitiveDependencies>
+      <includes>
+          <include>com.fasterxml.jackson.core:*:jar</include>
+      </includes>
+    </dependencySet>
+  </dependencySets>
+
+  <fileSets>
+
+    <fileSet>
+      <directory>misc/</directory>
+	  <outputDirectory>carbon</outputDirectory>
+      <includes>
+        <include>*.json</include>
+      </includes>
+    </fileSet>
+
+    <fileSet>
+      <directory>target</directory>
+      <outputDirectory>carbon</outputDirectory>
+	  <includes>
+        <include>carbondata-zeppelin-*.jar</include>
+      </includes>
+    </fileSet>
+
+  </fileSets>
+</assembly>
diff --git a/integration/zeppelin/misc/interpreter-setting.json b/integration/zeppelin/misc/interpreter-setting.json
@@ -0,0 +1,22 @@
+[
+  {
+    "group": "carbon",
+    "name": "carbon",
+    "className": "org.apache.carbondata.zeppelin.CarbonInterpreter",
+    "properties": {
+      "carbon.query.api.url": {
+        "envName": null,
+        "propertyName": "carbon.query.api.url",
+        "defaultValue": "",
+        "description": "API URL for request",
+        "type": "string"
+      }
+    },
+    "editor": {
+      "language": "sql",
+      "editOnDblClick": false,
+      "completionKey": "TAB",
+      "completionSupport": true
+    }
+  }
+]
diff --git a/integration/zeppelin/pom.xml b/integration/zeppelin/pom.xml
@@ -0,0 +1,99 @@
+<?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/xsd/maven-4.0.0.xsd">
+
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>org.apache.carbondata</groupId>
+    <artifactId>carbondata-parent</artifactId>
+    <version>1.5.0-SNAPSHOT</version>
+    <relativePath>../../pom.xml</relativePath>
+  </parent>
+
+  <artifactId>carbondata-zeppelin</artifactId>
+  <name>Apache CarbonData :: Zeppelin</name>
+
+  <properties>
+    <dev.path>${basedir}/../../dev</dev.path>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.zeppelin</groupId>
+      <artifactId>zeppelin-interpreter</artifactId>
+      <version>0.8.0</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.httpcomponents</groupId>
+      <artifactId>httpclient</artifactId>
+      <version>4.5.1</version>
+    </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-annotations</artifactId>
+      <version>2.8.0</version>
+    </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-databind</artifactId>
+      <version>2.8.11.1</version>
+    </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-core</artifactId>
+      <version>2.8.10</version>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <version>4.11</version>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <plugins>
+      <plugin>
+        <artifactId>maven-compiler-plugin</artifactId>
+        <configuration>
+          <source>1.8</source>
+          <target>1.8</target>
+        </configuration>
+      </plugin>
+
+	  <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-assembly-plugin</artifactId>
+        <configuration>
+          <descriptor>assembly/assembly.xml</descriptor>
+          <finalName>carbondata-zeppelin-${version}</finalName>
+        </configuration>
+        <executions>
+          <execution>
+            <phase>package</phase>
+            <goals>
+              <goal>single</goal>
+            </goals>
+          </execution>
+        </executions>
+     </plugin>
+
+    </plugins>
+  </build>
+</project>
diff --git a/integration/zeppelin/src/main/java/org/apache/carbondata/zeppelin/CarbonInterpreter.java b/integration/zeppelin/src/main/java/org/apache/carbondata/zeppelin/CarbonInterpreter.java
@@ -0,0 +1,186 @@
+/*
+ * 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.carbondata.zeppelin;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Arrays;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import org.apache.carbondata.zeppelin.response.CarbonResponse;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.http.HttpResponse;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.zeppelin.interpreter.Interpreter;
+import org.apache.zeppelin.interpreter.InterpreterContext;
+import org.apache.zeppelin.interpreter.InterpreterException;
+import org.apache.zeppelin.interpreter.InterpreterResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Carbon based interpreter for zeppelin
+ */
+public class CarbonInterpreter extends Interpreter {
+
+  public static final Logger logger = LoggerFactory.getLogger(CarbonInterpreter.class);
+
+  static final char TAB = '\t';
+
+  static final String LF = "\n";
+
+  /**
+   * Property which can be set in zeppelin to carbon REST API server
+   */
+  public static final String CARBON_QUERY_API_URL = "carbon.query.api.url";
+
+  /**
+   * These are the queries which need Table like output format
+   */
+  private static final String[] SEARCH_QUERIES = {"select", "list", "show", "desc"};
+
+  public CarbonInterpreter(Properties properties) {
+    super(properties);
+  }
+
+  @Override
+  public void open() throws InterpreterException {
+  }
+
+  @Override
+  public void close() throws InterpreterException {
+  }
+
+  @Override
+  public void cancel(InterpreterContext interpreterContext) throws InterpreterException {
+  }
+
+  @Override
+  public int getProgress(InterpreterContext interpreterContext) throws InterpreterException {
+    return 0;
+  }
+
+  @Override
+  public FormType getFormType() throws InterpreterException {
+    return FormType.SIMPLE;
+  }
+
+  @Override
+  public InterpreterResult interpret(String sql, InterpreterContext interpreterContext)
+          throws InterpreterException {
+    try {
+      return executeQuery.apply(sql);
+    } catch (RuntimeException e) {
+      logger.error("failed to query data in carbon ", e);
+      return new InterpreterResult(InterpreterResult.Code.ERROR, e.getMessage());
+    }
+  }
+
+  /**
+   * This will execute the given sql Query by sending a post request on CARBON_QUERY_API_URL
+   */
+  private Function<String, HttpResponse> doPost = sql -> {
+    // prepare the post body
+    String postContent = new StringBuilder("{\"sqlStatement\":")
+            .append("\"").append(sql).append("\"" + "}").toString();
+    logger.debug("post:" + postContent);
+
+    // prepare entity and set content type
+    StringEntity entity = new StringEntity(postContent, "UTF-8");
+    entity.setContentType("application/json; charset=UTF-8");
+
+    // get the POST url from interpreter property
+    String postURL = getProperty(CARBON_QUERY_API_URL);
+    logger.debug("post url:" + postURL);
+
+    // do POST and get response
+    HttpPost postRequest = new HttpPost(postURL);
+    postRequest.setEntity(entity);
+    HttpClient httpClient = HttpClientBuilder.create().build();
+    try {
+      return httpClient.execute(postRequest);
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  };
+
+  /**
+   * Check if output has to be sent as a able to zeppelin
+   */
+  private Function<String, Boolean> isTableFormatOutput = sql ->
+          (StringUtils.startsWithAny(sql, SEARCH_QUERIES));
+
+  /**
+   * returns InterpreterResult from CarbonResponse
+   */
+  private BiFunction<String, CarbonResponse, InterpreterResult> getResult = (sql, response) -> {
+    if (isTableFormatOutput.apply(sql.toLowerCase().trim())) {
+      //format only select queries and return as table
+      String formattedResult = Arrays
+              .stream(response.getRows())
+              .filter(Objects::nonNull)
+              .map(row -> StringUtils.join(row, TAB))
+              .collect(Collectors.joining(LF));
+      return new InterpreterResult(InterpreterResult.Code.SUCCESS, InterpreterResult.Type.TABLE,
+              formattedResult);
+    } else {
+      return new InterpreterResult(InterpreterResult.Code.SUCCESS, InterpreterResult.Type.TEXT,
+              response.getMessage());
+    }
+  };
+
+  /**
+   * Executes the given sql and return formatted result
+   */
+  private Function<String, InterpreterResult> executeQuery = sql -> {
+    try {
+      HttpResponse response = doPost.apply(sql);
+      // always close the content after reading fully to release connection
+      // IOUtils.toString will completely read the content
+      try (InputStream content = response.getEntity().getContent()) {
+        Optional<CarbonResponse> carbonResponse = CarbonResponse.parse(content);
+        int code = response.getStatusLine().getStatusCode();
+        if (code != 200) {
+          StringBuilder errorMessage = new StringBuilder("Failed : HTTP error code " + code + " .");
+          carbonResponse.ifPresent(rsp -> {
+            logger.error("Failed to execute query: " + rsp.getFullResponse());
+            errorMessage.append(rsp.getMessage());
+          });
+          return new InterpreterResult(InterpreterResult.Code.ERROR, errorMessage.toString());
+        } else {
+          return carbonResponse.map(rsp -> getResult.apply(sql, rsp))
+                  .orElseGet(() -> new InterpreterResult(InterpreterResult.Code.SUCCESS,
+                          InterpreterResult.Type.TEXT,
+                          "Query Success, but unable to parse response"));
+        }
+      }
+    } catch (IOException e) {
+      logger.error("Error executing query ", e);
+      return new InterpreterResult(InterpreterResult.Code.ERROR, e.getMessage());
+    }
+  };
+}