Skip to content

Initial support for CassandraSQLContext for %sql queries #86

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.

Sign up for GitHub

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

Closed
benbromhead wants to merge 10 commits into from
Closed

Initial support for CassandraSQLContext for %sql queries #86

Changes from 1 commit
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
fb27aba
Added support for CassandraSQLContext for %sql queries
bbromhead May 28, 2015
4808830
No longer set spark.cassandra props explicitly as this is done alread…
bbromhead Jul 1, 2015
f28a956
Merge branch 'master' into cassandraSQLsupport
bbromhead Jul 1, 2015
b400ee9
Adding integration tests for Cassandra SQL context
bbromhead Jul 9, 2015
7fef967
Minor style changes
bbromhead Jul 17, 2015
3f39f93
Changed SqarkInterpreter to only use the Hive context when explicitly…
bbromhead Jul 17, 2015
270dc3b
Merge branch 'master' of https://github.com/apache/incubator-zeppelin…
bbromhead Jul 17, 2015
4c8e638
Added CI test for Cassandra and Spark 1.3
bbromhead Jul 17, 2015
c453426
Reverted change to default spark context. Now uses HiveContext as per…
bbromhead Jul 22, 2015
056f6dd
turn off hive for Cassandra test
bbromhead Jul 22, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 53 additions & 33 deletions spark/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,11 +26,7 @@
import java.lang.reflect.Method;
import java.net.URL;
import java.net.URLClassLoader;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.*;

import org.apache.spark.HttpServer;
import org.apache.spark.SparkConf;
Expand Down Expand Up @@ -163,32 +159,6 @@ private boolean useHiveContext() {
return Boolean.parseBoolean(getProperty("zeppelin.spark.useHiveContext"));
}

public SQLContext getSQLContext() {
if (sqlc == null) {
if (useHiveContext()) {
String name = "org.apache.spark.sql.hive.HiveContext";
Constructor<?> hc;
try {
hc = getClass().getClassLoader().loadClass(name)
.getConstructor(SparkContext.class);
sqlc = (SQLContext) hc.newInstance(getSparkContext());
} catch (NoSuchMethodException | SecurityException
| ClassNotFoundException | InstantiationException
| IllegalAccessException | IllegalArgumentException
| InvocationTargetException e) {
logger.warn("Can't create HiveContext. Fallback to SQLContext", e);
// when hive dependency is not loaded, it'll fail.
// in this case SQLContext can be used.
sqlc = new SQLContext(getSparkContext());
}
} else {
sqlc = new SQLContext(getSparkContext());
}
}

return sqlc;
}

public DependencyResolver getDependencyResolver() {
if (dep == null) {
dep = new DependencyResolver(intp, sc, getProperty("zeppelin.dep.localrepo"));
Expand All @@ -213,6 +183,45 @@ private DepInterpreter getDepInterpreter() {
return null;
}

private boolean useCassandraContext() {
return Boolean.parseBoolean(getProperty("zeppelin.spark.useCassandraContext"));
}

private SQLContext loadCustomContext(final String contextName) {
Constructor<?> hc;
SQLContext context;
try {
hc = getClass().getClassLoader().loadClass(contextName)
.getConstructor(SparkContext.class);
context = (SQLContext) hc.newInstance(getSparkContext());
} catch (NoSuchMethodException | SecurityException
| ClassNotFoundException | InstantiationException
| IllegalAccessException | IllegalArgumentException
| InvocationTargetException e) {
logger.warn("Can't create " + contextName + ". Fallback to SQLContext", e);
// when hive dependency is not loaded, it'll fail.
// in this case SQLContext can be used.
context = new SQLContext(getSparkContext());
}
return context;
}

public SQLContext getSQLContext() {
if (sqlc == null) {
if (useCassandraContext()) {
sqlc = loadCustomContext("org.apache.spark.sql.cassandra.CassandraSQLContext");
logger.debug("Loading Cassandra SQL Context");
} else if (useHiveContext()) {
sqlc = loadCustomContext("org.apache.spark.sql.hive.HiveContext");
logger.debug("Loading Hive SQL Context");
} else {
sqlc = new SQLContext(getSparkContext());
logger.debug("Loading Standard SQL Context");
}
}
return sqlc;
}

public SparkContext createSparkContext() {
System.err.println("------ Create new SparkContext " + getProperty("master") + " -------");

Expand Down Expand Up @@ -240,12 +249,23 @@ public SparkContext createSparkContext() {
}
}

SparkConf conf =
new SparkConf()
SparkConf conf = new SparkConf()
.setMaster(getProperty("master"))
.setAppName(getProperty("spark.app.name"))
.set("spark.repl.class.uri", classServerUri);

if (useCassandraContext()) {
conf.set("spark.cassandra.connection.host", getProperty("spark.cassandra.connection.host"));

if (getProperty("spark.cassandra.auth.username") != null) {
conf.set("spark.cassandra.auth.username", getProperty("spark.cassandra.auth.username"));
}

if (getProperty("spark.cassandra.auth.password") != null) {
conf.set("spark.cassandra.auth.password", getProperty("spark.cassandra.auth.password"));
}
}

@Leemoonsoo Leemoonsoo Jun 8, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think line 268 sets all property into conf that property starts with 'spark.'. So isn't it duplicated?


if (jars.length > 0) {
conf.setJars(jars);
}
Expand Down