Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

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

Closed
wants to merge 7 commits into from
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
44 changes: 44 additions & 0 deletions 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```
37 changes: 37 additions & 0 deletions 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>
22 changes: 22 additions & 0 deletions 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
}
}
]
99 changes: 99 additions & 0 deletions 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>
@@ -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());
}
};
}