- - Version 2.0 + Version 3.0
- - Java Driver Version 3.0.0-rc1 + Java Driver Version 3.0.1
-
@@ -215,6 +215,14 @@
Cassandra version Documentation + 3.x ++ + http://docs.datastax.com/en/cql/3.3/cql/cqlIntro.html + + +2.2 @@ -464,6 +472,11 @@ @fetchSize=int value Apply the given fetch size to all queries in the paragraph + Request Timeout +@requestTimeOut=int value +Apply the given request timeout in millisecs to all queries in the paragraph +
@@ -506,6 +519,10 @@Fetch Size Any integer value ++ Request Timeout +Any integer value +
@@ -526,19 +543,19 @@ ); TRUNCATE spark_demo.ts; - # Timestamp in the past + // Timestamp in the past @timestamp=10 - # Force timestamp directly in the first insert + // Force timestamp directly in the first insert INSERT INTO spark_demo.ts(key,value) VALUES(1,'first insert') USING TIMESTAMP 100; - # Select some data to make the clock turn + // Select some data to make the clock turn SELECT * FROM spark_demo.albums LIMIT 100; - # Now insert using the timestamp parameter set at the beginning(10) + // Now insert using the timestamp parameter set at the beginning(10) INSERT INTO spark_demo.ts(key,value) VALUES(1,'second insert'); - # Check for the result. You should see 'first insert' + // Check for the result. You should see 'first insert' SELECT value FROM spark_demo.ts WHERE key=1; @@ -599,13 +616,13 @@@@ -615,10 +632,10 @@- @prepare[statement_name]=... + @prepare[statement-name]=... - @bind[statement_name]=’text’, 1223, ’2015-07-30 12:00:01’, null, true, [‘list_item1’, ’list_item2’] + @bind[statement-name]=’text’, 1223, ’2015-07-30 12:00:01’, null, true, [‘list_item1’, ’list_item2’] - @bind[statement_name_with_no_bound_value] + @bind[statement-name-with-no-bound-value] - @remove_prepare[statement_name] + @remove_prepare[statement-name]II @prepare
- You can use the syntax "@prepare[statement_name]=SELECT ..." to create a prepared statement. - The statement_name is mandatory because the interpreter prepares the given statement with the + You can use the syntax "@prepare[statement-name]=SELECT ..." to create a prepared statement. + The statement-name is mandatory because the interpreter prepares the given statement with the Java driver and saves the generated prepared statement in an internal map, using the provided - statement_name as search key. + statement-name as search key.
Please note that this internal prepared statement map is shared with all notebooks @@ -626,7 +643,7 @@
- If the interpreter encounters many @prepare for the same statement_name (key), + If the interpreter encounters many @prepare for the same statement-name (key), only the first statement will be taken into account.
@@ -645,7 +662,7 @@
For the above example, the prepared statement is "SELECT * FROM spark_demo.albums LIMIT ?". - "SELECT * FROM spark_demo.artists LIMIT ?" is ignored because an entry already exists in the + "SELECT * FROM spark_demo.artists LIMIT ?" is ignored because an entry already exists in the prepared statements map with the key select.
In the context of Zeppelin, a notebook can be scheduled to be executed at regular interval, @@ -712,7 +729,7 @@IV @remove_prepare
- To avoid for a prepared statement to stay forever in the prepared statement map, you can use the @remove_prepare[statement_name] syntax + To avoid for a prepared statement to stay forever in the prepared statement map, you can use the @remove_prepare[statement-name] syntax to remove it. Removing a non-existing prepared statement yields no error.
@@ -735,16 +752,22 @@@@ -766,13 +789,13 @@ In the above example, the first CQL query will be executed for performer='Sheryl Crow' AND style='Rock'. For subsequent queries, you can change the value directly using the form. Please note that we enclosed the {{ }} block between simple quotes (') because Cassandra expects a String here. - We could have also use the {{style='Rock'}} syntax but this time, the value + We could have also use the \${style='Rock'} syntax but this time, the value displayed on the form is 'Rock' and not Rock.- Instead of hard-coding your CQL queries, it is possible to use the mustache syntax ({{ }}) - to inject simple value or multiple choices forms. + Instead of hard-coding your CQL queries, it is possible to use + Zeppelin dynamic form + syntax to inject simple value or multiple choices forms. + + The legacy mustache syntax ( {{ }} ) to bind input text and select form is still supported but is deprecated and will be removed in future releases. +
+-- Legacy syntax --
The syntax for simple parameter is: {{input_Label=default value}}. The default value is mandatory because the first time the paragraph is executed, we launch the CQL query before rendering the form so at least one value should be provided.
The syntax for multiple choices parameter is: {{input_Label=value1 | value2 | … | valueN }}. By default the first choice is used for CQL query the first time the paragraph is executed. +-- End legacy syntax --
Example:
@@ -755,8 +778,8 @@ #Secondary index on performer style SELECT name, country, performer FROM spark_demo.performers - WHERE name='{{performer=Sheryl Crow|Doof|Fanfarlo|Los Paranoia}}' - AND styles CONTAINS '{{style=Rock}}'; + WHERE name='\${performer=Sheryl Crow|Doof|Fanfarlo|Los Paranoia}' + AND styles CONTAINS '\${style=Rock}';
It is also possible to use dynamic forms for prepared statements:@@ -892,7 +915,7 @@
- @bind[select]=='{{performer=Sheryl Crow|Doof|Fanfarlo|Los Paranoia}}', '{{style=Rock}}' + @bind[select]=='\${performer=Sheryl Crow|Doof|Fanfarlo|Los Paranoia}', '\${style=Rock}'cassandra.protocol.version -3 +4 cassandra.query.default.consistency @@ -947,20 +970,35 @@ @@ -974,6 +1012,20 @@+ 3.0 : +
+-
+
- Update documentation +
- Update interactive documentation +
- Add support for binary protocol V4 +
- Implement new
@requestTimeOutruntime option
+ - Upgrade Java driver version to 3.0.1 +
- Allow interpreter to add dynamic forms programmatically when using FormType.SIMPLE +
- Allow dynamic form using default Zeppelin syntax +
- Fixing typo on FallThroughPolicy +
- Look for data in AngularObjectRegistry before creating dynamic form +
- Add missing support for
ALTERstatements
+
-
diff --git a/cassandra/src/main/scala/org/apache/zeppelin/cassandra/InterpreterLogic.scala b/cassandra/src/main/scala/org/apache/zeppelin/cassandra/InterpreterLogic.scala
index 707c16a8bae..363da7b1d22 100644
--- a/cassandra/src/main/scala/org/apache/zeppelin/cassandra/InterpreterLogic.scala
+++ b/cassandra/src/main/scala/org/apache/zeppelin/cassandra/InterpreterLogic.scala
@@ -29,6 +29,7 @@ import com.datastax.driver.core._
import com.datastax.driver.core.exceptions.DriverException
import com.datastax.driver.core.policies.{LoggingRetryPolicy, FallthroughRetryPolicy, DowngradingConsistencyRetryPolicy, Policies}
import org.apache.zeppelin.cassandra.TextBlockHierarchy._
+import org.apache.zeppelin.display.AngularObjectRegistry
import org.apache.zeppelin.display.Input.ParamOption
import org.apache.zeppelin.interpreter.InterpreterResult.Code
import org.apache.zeppelin.interpreter.{InterpreterException, InterpreterResult, InterpreterContext}
@@ -41,17 +42,20 @@ import scala.collection.mutable.ArrayBuffer
/**
* Value object to store runtime query parameters
- * @param consistency consistency level
+ *
+ * @param consistency consistency level
* @param serialConsistency serial consistency level
* @param timestamp timestamp
* @param retryPolicy retry policy
* @param fetchSize query fetch size
+ * @param requestTimeOut request time out in millisecs
*/
case class CassandraQueryOptions(consistency: Option[ConsistencyLevel],
serialConsistency:Option[ConsistencyLevel],
timestamp: Option[Long],
retryPolicy: Option[RetryPolicy],
- fetchSize: Option[Int])
+ fetchSize: Option[Int],
+ requestTimeOut: Option[Int])
/**
* Singleton object to store constants
@@ -71,7 +75,7 @@ object InterpreterLogic {
val fallThroughRetryPolicy = FallthroughRetryPolicy.INSTANCE
val loggingDefaultRetryPolicy = new LoggingRetryPolicy(defaultRetryPolicy)
val loggingDownGradingRetryPolicy = new LoggingRetryPolicy(downgradingConsistencyRetryPolicy)
- val loggingFallThrougRetryPolicy = new LoggingRetryPolicy(fallThroughRetryPolicy)
+ val loggingFallThroughRetryPolicy = new LoggingRetryPolicy(fallThroughRetryPolicy)
val preparedStatements : mutable.Map[String,PreparedStatement] = new ConcurrentHashMap[String,PreparedStatement]().asScala
@@ -273,7 +277,13 @@ class InterpreterLogic(val session: Session) {
.flatMap(x => Option(x.value))
.headOption
- CassandraQueryOptions(consistency,serialConsistency, timestamp, retryPolicy, fetchSize)
+ val requestTimeOut: Option[Int] = parameters
+ .filter(_.paramType == RequestTimeOutParam)
+ .map(_.getParam[RequestTimeOut])
+ .flatMap(x => Option(x.value))
+ .headOption
+
+ CassandraQueryOptions(consistency,serialConsistency, timestamp, retryPolicy, fetchSize, requestTimeOut)
}
def generateSimpleStatement(st: SimpleStm, options: CassandraQueryOptions,context: InterpreterContext): SimpleStatement = {
@@ -305,19 +315,38 @@ class InterpreterLogic(val session: Session) {
def maybeExtractVariables(statement: String, context: InterpreterContext): String = {
+ def findInAngularRepository(variable: String): Option[AnyRef] = {
+ val registry = context.getAngularObjectRegistry
+ val noteId = context.getNoteId
+ val paragraphId = context.getParagraphId
+ val paragraphScoped: Option[AnyRef] = Option(registry.get(variable, noteId, paragraphId)).map[AnyRef](_.get())
+
+ paragraphScoped
+ }
+
def extractVariableAndDefaultValue(statement: String, exp: String):String = {
exp match {
- case MULTIPLE_CHOICES_VARIABLE_DEFINITION_PATTERN(variable,choices) => {
+ case MULTIPLE_CHOICES_VARIABLE_DEFINITION_PATTERN(variable, choices) => {
val escapedExp: String = exp.replaceAll( """\{""", """\\{""").replaceAll( """\}""", """\\}""").replaceAll("""\|""","""\\|""")
- val listChoices:List[String] = choices.trim.split(CHOICES_SEPARATOR).toList
- val paramOptions= listChoices.map(choice => new ParamOption(choice, choice))
- val selected = context.getGui.select(variable, listChoices.head, paramOptions.toArray)
- statement.replaceAll(escapedExp,selected.toString)
+ findInAngularRepository(variable) match {
+ case Some(value) => statement.replaceAll(escapedExp,value.toString)
+ case None => {
+ val listChoices:List[String] = choices.trim.split(CHOICES_SEPARATOR).toList
+ val paramOptions= listChoices.map(choice => new ParamOption(choice, choice))
+ val selected = context.getGui.select(variable, listChoices.head, paramOptions.toArray)
+ statement.replaceAll(escapedExp,selected.toString)
+ }
+ }
}
case SIMPLE_VARIABLE_DEFINITION_PATTERN(variable,defaultVal) => {
val escapedExp: String = exp.replaceAll( """\{""", """\\{""").replaceAll( """\}""", """\\}""")
- val value = context.getGui.input(variable,defaultVal)
- statement.replaceAll(escapedExp,value.toString)
+ findInAngularRepository(variable) match {
+ case Some(value) => statement.replaceAll(escapedExp,value.toString)
+ case None => {
+ val value = context.getGui.input(variable,defaultVal)
+ statement.replaceAll(escapedExp,value.toString)
+ }
+ }
}
case _ => throw new ParsingException(s"Invalid bound variable definition for '$exp' in '$statement'. It should be of form 'variable=defaultValue' or 'variable=value1|value2|...|valueN'")
}
@@ -336,10 +365,11 @@ class InterpreterLogic(val session: Session) {
case FallThroughRetryPolicy => statement.setRetryPolicy(fallThroughRetryPolicy)
case LoggingDefaultRetryPolicy => statement.setRetryPolicy(loggingDefaultRetryPolicy)
case LoggingDowngradingRetryPolicy => statement.setRetryPolicy(loggingDownGradingRetryPolicy)
- case LoggingFallThroughRetryPolicy => statement.setRetryPolicy(loggingFallThrougRetryPolicy)
+ case LoggingFallThroughRetryPolicy => statement.setRetryPolicy(loggingFallThroughRetryPolicy)
case _ => throw new InterpreterException(s"""Unknown retry policy ${options.retryPolicy.getOrElse("???")}""")
}
options.fetchSize.foreach(statement.setFetchSize(_))
+ options.requestTimeOut.foreach(statement.setReadTimeoutMillis(_))
}
private def createBoundStatement(codecRegistry: CodecRegistry, name: String, ps: PreparedStatement, rawBoundValues: String): BoundStatement = {
diff --git a/cassandra/src/main/scala/org/apache/zeppelin/cassandra/JavaDriverConfig.scala b/cassandra/src/main/scala/org/apache/zeppelin/cassandra/JavaDriverConfig.scala
index d64ad90c1d3..5b2dbefdc19 100644
--- a/cassandra/src/main/scala/org/apache/zeppelin/cassandra/JavaDriverConfig.scala
+++ b/cassandra/src/main/scala/org/apache/zeppelin/cassandra/JavaDriverConfig.scala
@@ -207,6 +207,16 @@ class JavaDriverConfig {
DEFAULT_MAX_REQUEST_PER_CONNECTION_LOCAL = "1024"
DEFAULT_MAX_REQUEST_PER_CONNECTION_REMOTE = "256"
return ProtocolVersion.V3
+ case "4" =>
+ DEFAULT_MAX_CONNECTION_PER_HOST_LOCAL = "1"
+ DEFAULT_MAX_CONNECTION_PER_HOST_REMOTE = "1"
+ DEFAULT_CORE_CONNECTION_PER_HOST_LOCAL = "1"
+ DEFAULT_CORE_CONNECTION_PER_HOST_REMOTE = "1"
+ DEFAULT_NEW_CONNECTION_THRESHOLD_LOCAL = "800"
+ DEFAULT_NEW_CONNECTION_THRESHOLD_REMOTE = "200"
+ DEFAULT_MAX_REQUEST_PER_CONNECTION_LOCAL = "1024"
+ DEFAULT_MAX_REQUEST_PER_CONNECTION_REMOTE = "256"
+ return ProtocolVersion.V4
case _ =>
DEFAULT_MAX_CONNECTION_PER_HOST_LOCAL = "1"
DEFAULT_MAX_CONNECTION_PER_HOST_REMOTE = "1"
diff --git a/cassandra/src/main/scala/org/apache/zeppelin/cassandra/ParagraphParser.scala b/cassandra/src/main/scala/org/apache/zeppelin/cassandra/ParagraphParser.scala
index e2cb64dbbe8..29c013f1171 100644
--- a/cassandra/src/main/scala/org/apache/zeppelin/cassandra/ParagraphParser.scala
+++ b/cassandra/src/main/scala/org/apache/zeppelin/cassandra/ParagraphParser.scala
@@ -44,6 +44,7 @@ object ParagraphParser {
LOGGING_DEFAULT_RETRY, LOGGING_DOWNGRADING_RETRY, LOGGING_FALLTHROUGH_RETRY)
.mkString("""^\s*@retryPolicy\s*=\s*(""", "|" , """)\s*$""").r
val FETCHSIZE_PATTERN = """^\s*@fetchSize\s*=\s*([0-9]+)\s*$""".r
+ val REQUEST_TIMEOUT_PATTERN = """^\s*@requestTimeOut\s*=\s*([0-9]+)\s*$""".r
val SIMPLE_STATEMENT_PATTERN = """([^;]+;)""".r
val PREPARE_STATEMENT_PATTERN = """^\s*@prepare\[([^]]+)\]\s*=\s*([^;]+)$""".r
@@ -69,7 +70,7 @@ object ParagraphParser {
val UDF_PATTERN = """(?is)\s*(CREATE(?:\s+OR REPLACE)?\s+FUNCTION(?:\s+IF\s+NOT\s+EXISTS)?.+?(?:\s+|\n|\r|\f)AS(?:\s+|\n|\r|\f)(?:'|\$\$).+?(?:'|\$\$)\s*;)""".r
val GENERIC_STATEMENT_PREFIX =
- """(?is)\s*(?:INSERT|UPDATE|DELETE|SELECT|CREATE|UPDATE|
+ """(?is)\s*(?:INSERT|UPDATE|DELETE|SELECT|CREATE|ALTER|
|DROP|GRANT|REVOKE|TRUNCATE|LIST|USE)\s+""".r
val VALID_IDENTIFIER = "[a-z][a-z0-9_]*"
@@ -146,6 +147,7 @@ class ParagraphParser extends RegexParsers{
def timestamp: Parser[Timestamp] = """\s*@timestamp.+""".r ^^ {case x => extractTimestamp(x.trim)}
def retryPolicy: Parser[RetryPolicy] = """\s*@retryPolicy.+""".r ^^ {case x => extractRetryPolicy(x.trim)}
def fetchSize: Parser[FetchSize] = """\s*@fetchSize.+""".r ^^ {case x => extractFetchSize(x.trim)}
+ def requestTimeOut: Parser[RequestTimeOut] = """\s*@requestTimeOut.+""".r ^^ {case x => extractRequestTimeOut(x.trim)}
//Statements
def createFunctionStatement: Parser[SimpleStm] = UDF_PATTERN ^^{case x => extractUdfStatement(x.trim)}
@@ -188,7 +190,7 @@ class ParagraphParser extends RegexParsers{
case begin ~ cqls ~ end => BatchStm(extractBatchType(begin),cqls)}
def queries:Parser[List[AnyBlock]] = rep(singleLineComment | multiLineComment | consistency | serialConsistency |
- timestamp | retryPolicy | fetchSize | removePrepare | prepare | bind | batch | describeCluster |
+ timestamp | retryPolicy | fetchSize | requestTimeOut | removePrepare | prepare | bind | batch | describeCluster |
describeKeyspace | describeKeyspaces |
describeTable | describeTables |
describeType | describeTypes |
@@ -244,6 +246,14 @@ class ParagraphParser extends RegexParsers{
}
}
+ def extractRequestTimeOut(text: String): RequestTimeOut = {
+ text match {
+ case REQUEST_TIMEOUT_PATTERN(requestTimeOut) => RequestTimeOut(requestTimeOut.trim.toInt)
+ case _ => throw new InterpreterException(s"Invalid syntax for @requestTimeOut. " +
+ s"It should comply to the pattern ${REQUEST_TIMEOUT_PATTERN.toString}")
+ }
+ }
+
def extractSimpleStatement(text: String): SimpleStm = {
text match {
case SIMPLE_STATEMENT_PATTERN(statement) => SimpleStm(statement)
diff --git a/cassandra/src/main/scala/org/apache/zeppelin/cassandra/TextBlockHierarchy.scala b/cassandra/src/main/scala/org/apache/zeppelin/cassandra/TextBlockHierarchy.scala
index 61a2d8d9cb5..be55564b316 100644
--- a/cassandra/src/main/scala/org/apache/zeppelin/cassandra/TextBlockHierarchy.scala
+++ b/cassandra/src/main/scala/org/apache/zeppelin/cassandra/TextBlockHierarchy.scala
@@ -44,6 +44,7 @@ object TextBlockHierarchy {
object TimestampParam extends ParameterType
object RetryPolicyParam extends ParameterType
object FetchSizeParam extends ParameterType
+ object RequestTimeOutParam extends ParameterType
abstract class QueryParameters(val paramType: ParameterType) extends AnyBlock(ParameterBlock) {
@@ -60,6 +61,8 @@ object TextBlockHierarchy {
case class FetchSize(value: Int) extends QueryParameters(FetchSizeParam)
+ case class RequestTimeOut(value: Int) extends QueryParameters(RequestTimeOutParam)
+
abstract class RetryPolicy extends QueryParameters(RetryPolicyParam)
object DefaultRetryPolicy extends RetryPolicy
diff --git a/cassandra/src/test/java/org/apache/zeppelin/cassandra/CassandraInterpreterTest.java b/cassandra/src/test/java/org/apache/zeppelin/cassandra/CassandraInterpreterTest.java
index 560c57e8239..db3c3914683 100644
--- a/cassandra/src/test/java/org/apache/zeppelin/cassandra/CassandraInterpreterTest.java
+++ b/cassandra/src/test/java/org/apache/zeppelin/cassandra/CassandraInterpreterTest.java
@@ -35,6 +35,7 @@
import info.archinnov.achilles.embedded.CassandraEmbeddedServerBuilder;
+import org.apache.zeppelin.display.AngularObjectRegistry;
import org.apache.zeppelin.interpreter.Interpreter;
import org.apache.zeppelin.interpreter.InterpreterContext;
import org.apache.zeppelin.interpreter.InterpreterResult;
@@ -45,7 +46,6 @@
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
-import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
@@ -63,7 +63,7 @@ public class CassandraInterpreterTest {
.withScript("prepare_data.cql")
.withProtocolVersion(ProtocolVersion.V3)
.buildNativeSessionOnly();
-// public static Session session = null;
+
private static CassandraInterpreter interpreter;
@Mock(answer = Answers.RETURNS_DEEP_STUBS)
@@ -73,7 +73,7 @@ public class CassandraInterpreterTest {
public static void setUp() {
Properties properties = new Properties();
final Cluster cluster = session.getCluster();
-// final Cluster cluster = null;
+
properties.setProperty(CASSANDRA_CLUSTER_NAME, cluster.getClusterName());
properties.setProperty(CASSANDRA_COMPRESSION_PROTOCOL, "NONE");
properties.setProperty(CASSANDRA_CREDENTIALS_USERNAME, "none");
@@ -289,6 +289,19 @@ public void should_execute_statement_with_retry_policy() throws Exception {
assertThat(actual.code()).isEqualTo(Code.SUCCESS);
}
+ @Test
+ public void should_execute_statement_with_request_timeout() throws Exception {
+ //Given
+ String statement = "@requestTimeOut=10000000\n" +
+ "SELECT * FROM zeppelin.artists;";
+
+ //When
+ final InterpreterResult actual = interpreter.interpret(statement, intrContext);
+
+ //Then
+ assertThat(actual.code()).isEqualTo(Code.SUCCESS);
+ }
+
@Test
public void should_execute_prepared_and_bound_statements() throws Exception {
//Given
@@ -354,6 +367,8 @@ public void should_exception_when_executing_unknown_bound_statement() throws Exc
@Test
public void should_extract_variable_from_statement() throws Exception {
//Given
+ AngularObjectRegistry angularObjectRegistry = new AngularObjectRegistry("cassandra", null);
+ when(intrContext.getAngularObjectRegistry()).thenReturn(angularObjectRegistry);
when(intrContext.getGui().input("login", "hsue")).thenReturn("hsue");
when(intrContext.getGui().input("age", "27")).thenReturn("27");
diff --git a/cassandra/src/test/java/org/apache/zeppelin/cassandra/InterpreterLogicTest.java b/cassandra/src/test/java/org/apache/zeppelin/cassandra/InterpreterLogicTest.java
index 77b5cd8b445..698397ae946 100644
--- a/cassandra/src/test/java/org/apache/zeppelin/cassandra/InterpreterLogicTest.java
+++ b/cassandra/src/test/java/org/apache/zeppelin/cassandra/InterpreterLogicTest.java
@@ -31,6 +31,8 @@
import com.datastax.driver.core.Session;
import com.datastax.driver.core.SimpleStatement;
import com.datastax.driver.core.Statement;
+
+import org.apache.zeppelin.display.AngularObjectRegistry;
import org.apache.zeppelin.display.GUI;
import org.apache.zeppelin.display.Input.ParamOption;
import org.apache.zeppelin.interpreter.InterpreterContext;
@@ -101,6 +103,8 @@ public void should_exception_while_parsing_input() throws Exception {
@Test
public void should_extract_variable_and_default_value() throws Exception {
//Given
+ AngularObjectRegistry angularObjectRegistry = new AngularObjectRegistry("cassandra", null);
+ when(intrContext.getAngularObjectRegistry()).thenReturn(angularObjectRegistry);
when(intrContext.getGui().input("table", "zeppelin.demo")).thenReturn("zeppelin.demo");
when(intrContext.getGui().input("id", "'John'")).thenReturn("'John'");
@@ -114,6 +118,8 @@ public void should_extract_variable_and_default_value() throws Exception {
@Test
public void should_extract_variable_and_choices() throws Exception {
//Given
+ AngularObjectRegistry angularObjectRegistry = new AngularObjectRegistry("cassandra", null);
+ when(intrContext.getAngularObjectRegistry()).thenReturn(angularObjectRegistry);
when(intrContext.getGui().select(eq("name"), eq("'Paul'"), optionsCaptor.capture())).thenReturn("'Jack'");
//When
@@ -141,6 +147,23 @@ public void should_extract_no_variable() throws Exception {
assertThat(actual).isEqualTo("SELECT * FROM zeppelin.demo");
}
+ @Test
+ public void should_extract_variable_from_angular_object_registry() throws Exception {
+ //Given
+ AngularObjectRegistry angularObjectRegistry = new AngularObjectRegistry("cassandra", null);
+ angularObjectRegistry.add("id", "from_angular_registry", "noteId", "paragraphId");
+ when(intrContext.getAngularObjectRegistry()).thenReturn(angularObjectRegistry);
+ when(intrContext.getNoteId()).thenReturn("noteId");
+ when(intrContext.getParagraphId()).thenReturn("paragraphId");
+
+ //When
+ final String actual = helper.maybeExtractVariables("SELECT * FROM zeppelin.demo WHERE id='{{id=John}}'", intrContext);
+
+ //Then
+ assertThat(actual).isEqualTo("SELECT * FROM zeppelin.demo WHERE id='from_angular_registry'");
+ verify(intrContext, never()).getGui();
+ }
+
@Test
public void should_error_if_incorrect_variable_definition() throws Exception {
//Given
@@ -203,6 +226,18 @@ public void should_extract_retry_policy_option() throws Exception {
assertThat(actual.retryPolicy().get()).isSameAs(DowngradingRetryPolicy$.MODULE$);
}
+ @Test
+ public void should_extract_request_timeout_option() throws Exception {
+ //Given
+ List
- many query parameters can be set in the same paragraph
- if the same query parameter is set many time with different values,the interpreter only take into account the first value
- each query parameter applies to all CQL statement in the same paragraph,unless you override the option using plain CQL text (like forcing timestamp with the USING clause)
- the order of each query parameter with regard to CQL statement does not matter
- @prepare
- @bind
- @remove_prepared
- String values should be enclosed between simple quotes ( ‘ )
- Date values should be enclosed between simple quotes ( ‘ ) and respect the formats:
- yyyy-MM-dd HH:MM:ss
- yyyy-MM-dd HH:MM:ss.SSS
- null is parsed as-is
- boolean (true|false) are parsed as-is
- collection values must follow thestandard CQL syntax:
- list: [‘list_item1’, ’list_item2’, ...]
- set: {‘set_item1’, ‘set_item2’, …}
- map: {‘key1’: ‘val1’, ‘key2’: ‘val2’, …}
- tuple values should be enclosed between parenthesis(see tuple CQL syntax):(‘text’, 123, true)
- udt values should be enclosed between brackets(see udt CQL syntax):{stree_name: ‘Beverly Hills’, number: 104, zip_code: 90020, state: ‘California’, …}
- First go to the Interpreter menu and click on the Create button
- In the interpreter creation form, put cass-instance2 as Name and select the cassandra in the interpreter drop-down list
- Click on Save to create the new interpreter instance. Now you should be able to see it in the interpreter list
- Go back to your notebook and click on the Gear icon to configure interpreter bindings. You should be able to see and select the cass-instance2 interpreter instance in the available interpreter list instead of the standard cassandra instance
- Update help menu and add changelog
- Add Support for User Defined Functions, User Defined Aggregates and Materialized Views
- Upgrade Java driver version to 3.0.0-rc1
- Initial version
- many query parameters can be set in the same paragraph
- if the same query parameter is set many time with different values,the interpreter only take into account the first value
- each query parameter applies to all CQL statement in the same paragraph,unless you override the option using plain CQL text (like forcing timestamp with the USING clause)
- the order of each query parameter with regard to CQL statement does not matter
- @prepare
- @bind
- @remove_prepared
- String values should be enclosed between simple quotes ( ‘ )
- Date values should be enclosed between simple quotes ( ‘ ) and respect the formats:
- yyyy-MM-dd HH:MM:ss
- yyyy-MM-dd HH:MM:ss.SSS
- null is parsed as-is
- boolean (true|false) are parsed as-is
- collection values must follow thestandard CQL syntax:
- list: [‘list_item1’, ’list_item2’, ...]
- set: {‘set_item1’, ‘set_item2’, …}
- map: {‘key1’: ‘val1’, ‘key2’: ‘val2’, …}
- tuple values should be enclosed between parenthesis(see tuple CQL syntax):(‘text’, 123, true)
- udt values should be enclosed between brackets(see udt CQL syntax):{stree_name: ‘Beverly Hills’, number: 104, zip_code: 90020, state: ‘California’, …}
- shared : same JVM and same Interpreter instance for all notes
- scoped : same JVM but different Interpreter instances, one for each note
- isolated : different JVM running a single Interpreter instance, one JVM for each note
- Update documentation
- Update interactive documentation
- Add support for binary protocol V4
- Implement new
@requestTimeOutruntime option - Upgrade Java driver version to 3.0.1
- Allow interpreter to add dynamic forms programmatically when using FormType.SIMPLE
- Allow dynamic form using default Zeppelin syntax
- Fixing typo on FallThroughPolicy
- Look for data in AngularObjectRegistry before creating dynamic form
- Add missing support for
ALTERstatements - Update help menu and add changelog
- Add Support for User Defined Functions, User Defined Aggregates and Materialized Views
- Upgrade Java driver version to 3.0.0-rc1
- Initial version
options = Arrays. asList(new RequestTimeOut(100)); + + //When + final CassandraQueryOptions actual = helper.extractQueryOptions(toScalaList(options)); + + //Then + assertThat(actual.requestTimeOut().get()).isEqualTo(100); + } + @Test public void should_generate_simple_statement() throws Exception { //Given @@ -211,6 +246,7 @@ public void should_generate_simple_statement() throws Exception { Option. empty(), Option.empty(), Option. empty(), + Option.empty(), Option.empty()); //When @@ -232,6 +268,7 @@ public void should_generate_batch_statement() throws Exception { Option. empty(), Option.empty(), Option. empty(), + Option.empty(), Option.empty()); //When diff --git a/cassandra/src/test/resources/scalate/Help.html b/cassandra/src/test/resources/scalate/Help.html index 1c82c66dd13..d784fe3dfb2 100644 --- a/cassandra/src/test/resources/scalate/Help.html +++ b/cassandra/src/test/resources/scalate/Help.html @@ -1 +1 @@ - \ No newline at end of file +I CQL Statements
This interpreter is compatible with any CQL statement supported by Cassandra. Ex:
INSERT INTO users(login,name) VALUES('jdoe','John DOE');SELECT * FROM users WHERE login='jdoe';
Each statement should be separated by a semi-colon (;).
Multi-line statements as well as multiple statements on the same lineare also supported as long as they are separated by a semi-colon. Ex:USE spark_demo;SELECT * FROM albums_by_country LIMIT 1; SELECT * FROM countries LIMIT 1;SELECT *FROM artistsWHERE login='jlennon';
Batch statements are supported and can span multiple lines, as well asDDL(CREATE/ALTER/DROP) statements:BEGIN BATCHINSERT INTO users(login,name) VALUES('jdoe','John DOE');INSERT INTO users_preferences(login,account_type) VALUES('jdoe','BASIC');APPLY BATCH;CREATE TABLE IF NOT EXISTS test(key int PRIMARY KEY,value text);
CQL statements are case-insensitive (except for column names and values).This means that the following statements are equivalent and valid:INSERT INTO users(login,name) VALUES('jdoe','John DOE');Insert into users(login,name) vAlues('hsue','Helen SUE');
The complete list of all CQL statements and versions can be found below:Cassandra version Documentation 2.2 http://docs.datastax.com/en/cql/3.3/cql/cqlIntro.html 2.1 & 2.0 http://docs.datastax.com/en/cql/3.1/cql/cql_intro_c.html 1.2 http://docs.datastax.com/en/cql/3.0/cql/aboutCQL.html II Comments
It is possible to add comments between statements. Single line comments start with thehash sign (#) or double slashes (//). Multi-line comments are enclosed between/** and **/. Ex:
#Single line comment style 1INSERT INTO users(login,name) VALUES('jdoe','John DOE');//Single line comment style 2/**Multi linecomments**/Insert into users(login,name) vAlues('hsue','Helen SUE');III Syntax Validation
The interpreters is shipped with a built-in syntax validator. This validator onlychecks for basic syntax errors. All CQL-related syntax validation is delegateddirectly to Cassandra
Most of the time, syntax errors are due to missing semi-colons between statements or typo errors.I Commands For Discovery
To make schema discovery easier and more interactive, the following commands are supported:
Command Description DESCRIBE CLUSTER; Show the current cluster name and its partitioner DESCRIBE KEYSPACES; List all existing keyspaces in the cluster and their configuration(replication factor, durable write ...) DESCRIBE TABLES; List all existing keyspaces in the cluster and for each, all the tables name DESCRIBE TYPES; List all existing keyspaces in the cluster and for each, all the types name DESCRIBE FUNCTIONS; List all existing keyspaces in the cluster and for each, all the functions name and arguments DESCRIBE AGGREGATES; List all existing keyspaces in the cluster and for each, all the aggregates name and arguments DESCRIBE MATERIALIZED VIEWS; List all existing keyspaces in the cluster and for each, all the materialized view name DESCRIBE KEYSPACE <keyspace name>; Describe the given keyspace configuration and all its table details (name, columns, ...) DESCRIBE TABLE (<keyspace name>).<table name>; Describe the given table. If the keyspace is not provided, the currentlogged in keyspace is used. If there is no logged in keyspace,the default system keyspace is used. If no table is found, an error message is raised DESCRIBE TYPE (<keyspace name>).<type name>; Describe the given type(UDT). If the keyspace is not provided, the currentlogged in keyspace is used. If there is no logged in keyspace,the default system keyspace is used. If no type is found, an error message is raised DESCRIBE FUNCTION (<keyspace name>).<function name>; Describe the given function. If the keyspace is not provided, the currentlogged in keyspace is used. If there is no logged in keyspace,the default system keyspace is used. If no function is found, an error message is raised DESCRIBE AGGREGATE (<keyspace name>).<aggregate name>; Describe the given aggregate. If the keyspace is not provided, the currentlogged in keyspace is used. If there is no logged in keyspace,the default system keyspace is used. If no aggregate is found, an error message is raised DESCRIBE MATERIALIZED VIEW (<keyspace name>).<view name>; Describe the given materialized view. If the keyspace is not provided, the currentlogged in keyspace is used. If there is no logged in keyspace,the default system keyspace is used. If no materialized view is found, an error message is raised Please note that each DESCRIBE command should be ended by a semi-colon.II Schema Display
The schema objects (cluster, keyspace, table, type, view, function & aggregate) are displayed in a tabular format. There is a drop-down menu on the top left corner to expand objects details. On the top right menu is shown the Icon legend.
Sometimes you want to be able to pass runtime query parameters to your statements.Those parameters are not part of the CQL specs and are specific to the interpreter.Below is the list of all parameters:
Query Parameters
Parameter Syntax Description Consistency Level @consistency=value Apply the given consistency level to all queries in the paragraph Serial Consistency Level @serialConsistency=value Apply the given serial consistency level to all queries in the paragraph Timestamp @timestamp=long value Apply the given timestamp to all queries in the paragraph.
Please note that timestamp value passed directly in CQL statement will override this valueRetry Policy @retryPolicy=value Apply the given retry policy to all queries in the paragraph Fetch Size @fetchSize=int value Apply the given fetch size to all queries in the paragraph
Some parameters only accept restricted values:Allowed Values
Parameter Possible Values Consistency Level ALL, ANY, ONE, TWO, THREE, QUORUM, LOCAL_ONE, LOCAL_QUORUM, EACH_QUORUM Serial Consistency Level SERIAL, LOCAL_SERIAL Timestamp Any long value Retry Policy DEFAULT, DOWNGRADING_CONSISTENCY, FALLTHROUGH, LOGGING_DEFAULT,LOGGING_DOWNGRADING, LOGGING_FALLTHROUGH Fetch Size Any integer value Please note that you should not add semi-colon (;) at the end of each parameter statementSome example:CREATE TABLE IF NOT EXISTS spark_demo.ts(key int PRIMARY KEY,value text);TRUNCATE spark_demo.ts;# Timestamp in the past@timestamp=10# Force timestamp directly in the first insertINSERT INTO spark_demo.ts(key,value) VALUES(1,'first insert') USING TIMESTAMP 100;# Select some data to make the clock turnSELECT * FROM spark_demo.albums LIMIT 100;# Now insert using the timestamp parameter set at the beginning(10)INSERT INTO spark_demo.ts(key,value) VALUES(1,'second insert');# Check for the result. You should see 'first insert'SELECT value FROM spark_demo.ts WHERE key=1;
Some remarks about query parameters:I Syntax
For performance reason, it is better to prepare statements before-hand and reusethem later by providing bound values. This interpreter provides 3 commands to handle prepared andbound statements:
Example:@prepare[statement_name]=...@bind[statement_name]=’text’, 1223, ’2015-07-30 12:00:01’, null, true, [‘list_item1’, ’list_item2’]@bind[statement_name_with_no_bound_value]@remove_prepare[statement_name]
II @prepare
You can use the syntax "@prepare[statement_name]=SELECT ..." to create a prepared statement.The statement_name is mandatory because the interpreter prepares the given statement with theJava driver and saves the generated prepared statement in an internal map, using the providedstatement_name as search key.
Please note that this internal prepared statement map is shared with all notebooksand all paragraphs because there is only one instance of the interpreter for CassandraIf the interpreter encounters many @prepare for the same statement_name (key),only the first statement will be taken into account.
Example:@prepare[select]=SELECT * FROM spark_demo.albums LIMIT ?@prepare[select]=SELECT * FROM spark_demo.artists LIMIT ?
For the above example, the prepared statement is "SELECT * FROM spark_demo.albums LIMIT ?"."SELECT * FROM spark_demo.artists LIMIT ?" is ignored because an entry already exists in theprepared statements map with the key select.
In the context of Zeppelin, a notebook can be scheduled to be executed at regular interval,thus it is necessary to avoid re-preparing many time the same statement (considered an anti-pattern).III @bind
Once the statement is prepared (possibly in a separated notebook/paragraph). You can bind values to it:
@bind[select_first]=10
Bound values are not mandatory for the @bind statement.However if you provide bound values, they need to comply to some syntax:It is possible to use the @bind statement inside a batch:BEGIN BATCH@bind[insert_user]='jdoe','John DOE'UPDATE users SET age = 27 WHERE login='hsue';APPLY BATCH;
IV @remove_prepare
To avoid for a prepared statement to stay forever in the prepared statement map, you can use the @remove_prepare[statement_name] syntaxto remove it. Removing a non-existing prepared statement yields no error.
Instead of hard-coding your CQL queries, it is possible to use the mustache syntax ({{ }})to inject simple value or multiple choices forms.
The syntax for simple parameter is: {{input_Label=default value}}.The default value is mandatory because the first time the paragraph is executed,we launch the CQL query before rendering the form so at least one value should be provided.
The syntax for multiple choices parameter is: {{input_Label=value1 | value2 | … | valueN }}.By default the first choice is used for CQL query the first time the paragraph is executed.
Example:#Secondary index on performer styleSELECT name, country, performerFROM spark_demo.performersWHERE name='{{performer=Sheryl Crow|Doof|Fanfarlo|Los Paranoia}}'AND styles CONTAINS '{{style=Rock}}';
In the above example, the first CQL query will be executed for performer='Sheryl Crow'AND style='Rock'. For subsequent queries, you can change the value directly using the form.Please note that we enclosed the {{ }} block between simple quotes (') because Cassandra expects a String here.We could have also use the {{style='Rock'}} syntax but this time, the valuedisplayed on the form is 'Rock' and not Rock.It is also possible to use dynamic forms for prepared statements:
@bind[select]=='{{performer=Sheryl Crow|Doof|Fanfarlo|Los Paranoia}}', '{{style=Rock}}'The Cassandra interpreter comes with some some configuration values for the Java driver:Interpreter Configuration
Parameter Default Value cassandra.cluster Test Cluster cassandra.compression.protocol NONE, possible values: LZ4, SNAPPY cassandra.credentials.password none cassandra.credentials.username none cassandra.hosts localhost cassandra.interpreter.parallelism 10 cassandra.keyspace system cassandra.load.balancing.policy DEFAULT, or a FQCN of a custom class cassandra.max.schema.agreement.wait.second 10 cassandra.native.port 9042 cassandra.pooling.core.connection.per.host.local Protocol V2 and below: 2, V3 and above: 1 cassandra.pooling.core.connection.per.host.remote Protocol V2 and below: 1, V3 and above: 1 cassandra.pooling.heartbeat.interval.seconds 30 cassandra.pooling.idle.timeout.seconds Test Cluster cassandra.pooling.max.connection.per.host.local Protocol V2 and below: 8, V3 and above: 1 cassandra.pooling.max.connection.per.host.remote Protocol V2 and below: 2, V3 and above: 1 cassandra.pooling.max.request.per.connection.local Protocol V2 and below: 128, V3 and above: 1024 cassandra.pooling.max.request.per.connection.remote Protocol V2 and below: 128, V3 and above: 256 cassandra.pooling.new.connection.threshold.local Protocol V2 and below: 100, V3 and above: 800 cassandra.pooling.new.connection.threshold.remote Protocol V2 and below: 100, V3 and above: 200 cassandra.pooling.pool.timeout.millisecs 5000 cassandra.protocol.version 3 cassandra.query.default.consistency ONE cassandra.query.default.fetchSize 5000 cassandra.query.default.serial.consistency SERIAL cassandra.reconnection.policy DEFAULT, or a FQCN of a custom class cassandra.retry.policy DEFAULT, or a FQCN of a custom class cassandra.socket.connection.timeout.millisecs 500 cassandra.socket.read.timeout.millisecs 12000 cassandra.socket.tcp.no_delay true cassandra.speculative.execution.policy DEFAULT, or a FQCN of a custom class It is possible to execute many paragraphs in parallel. However, at the back-end side, we’re still using synchronous queries. Asynchronous execution is only possible when it is possible to return a Future value in the InterpreterResult. It may be an interesting proposal for the Zeppelin project.
Another caveat is that the same com.datastax.driver.core.Session object is used for all notebooks and paragraphs. Consequently, if you use the USE keyspace name; statement to log into a keyspace, it will change the keyspace for all current users of the Cassandra interpreter because we only create 1 com.datastax.driver.core.Session object per instance of Cassandra interpreter.
The same remark does apply to the prepared statement hash map, it is shared by all users using the same instance of Cassandra interpreter.
Until Zeppelin offers a real multi-users separation, there is a work-around to segregate user environment and states: create different Cassandra interpreter instances2.0 :If you encounter a bug for this interpreter, please create a JIRA ticket and ping me on Twitter at @doanduyhai\ No newline at end of file diff --git a/cassandra/src/test/scala/org/apache/zeppelin/cassandra/ParagraphParserTest.scala b/cassandra/src/test/scala/org/apache/zeppelin/cassandra/ParagraphParserTest.scala index 520f7a61fc5..4c5c92934e4 100644 --- a/cassandra/src/test/scala/org/apache/zeppelin/cassandra/ParagraphParserTest.scala +++ b/cassandra/src/test/scala/org/apache/zeppelin/cassandra/ParagraphParserTest.scala @@ -182,6 +182,12 @@ class ParagraphParserTest extends FlatSpec parsed should matchPattern { case parser.Success(FetchSize(100), _) =>} } + "Parser" should "parse request timeout" in { + val query:String ="@requestTimeOut=100" + val parsed = parser.parseAll(parser.requestTimeOut, query) + parsed should matchPattern { case parser.Success(RequestTimeOut(100), _) =>} + } + "Parser" should "fails parsing invalid fetch size" in { val query:String =""" @fetchSize=TEST""".stripMargin val ex = intercept[InterpreterException] { @@ -944,4 +950,15 @@ class ParagraphParserTest extends FlatSpec case parser.Success(List(SimpleStm(query)), _) => } } + + "Parser" should "parse ALTER KEYSPACE" in { + val query = "ALTER KEYSPACE toto WITH replication = " + + "{'class': 'SimpleStrategy', 'replication_factor': 1};" + + val parsed = parser.parseAll(parser.queries, query) + + parsed should matchPattern { + case parser.Success(List(SimpleStm(query)), _) => + } + } } diff --git a/conf/zeppelin-site.xml.template b/conf/zeppelin-site.xml.template index 2d1fea1ba36..11ad8660cd9 100755 --- a/conf/zeppelin-site.xml.template +++ b/conf/zeppelin-site.xml.template @@ -178,7 +178,7 @@I CQL Statements
This interpreter is compatible with any CQL statement supported by Cassandra. Ex:
INSERT INTO users(login,name) VALUES('jdoe','John DOE');SELECT * FROM users WHERE login='jdoe';
Each statement should be separated by a semi-colon (;).
Multi-line statements as well as multiple statements on the same lineare also supported as long as they are separated by a semi-colon. Ex:USE spark_demo;SELECT * FROM albums_by_country LIMIT 1; SELECT * FROM countries LIMIT 1;SELECT *FROM artistsWHERE login='jlennon';
Batch statements are supported and can span multiple lines, as well asDDL(CREATE/ALTER/DROP) statements:BEGIN BATCHINSERT INTO users(login,name) VALUES('jdoe','John DOE');INSERT INTO users_preferences(login,account_type) VALUES('jdoe','BASIC');APPLY BATCH;CREATE TABLE IF NOT EXISTS test(key int PRIMARY KEY,value text);
CQL statements are case-insensitive (except for column names and values).This means that the following statements are equivalent and valid:INSERT INTO users(login,name) VALUES('jdoe','John DOE');Insert into users(login,name) vAlues('hsue','Helen SUE');
The complete list of all CQL statements and versions can be found below:II Comments
It is possible to add comments between statements. Single line comments start with thehash sign (#) or double slashes (//). Multi-line comments are enclosed between/** and **/. Ex:
#Single line comment style 1INSERT INTO users(login,name) VALUES('jdoe','John DOE');//Single line comment style 2/**Multi linecomments**/Insert into users(login,name) vAlues('hsue','Helen SUE');III Syntax Validation
The interpreters is shipped with a built-in syntax validator. This validator onlychecks for basic syntax errors. All CQL-related syntax validation is delegateddirectly to Cassandra
Most of the time, syntax errors are due to missing semi-colons between statements or typo errors.I Commands For Discovery
To make schema discovery easier and more interactive, the following commands are supported:
Command Description DESCRIBE CLUSTER; Show the current cluster name and its partitioner DESCRIBE KEYSPACES; List all existing keyspaces in the cluster and their configuration(replication factor, durable write ...) DESCRIBE TABLES; List all existing keyspaces in the cluster and for each, all the tables name DESCRIBE TYPES; List all existing keyspaces in the cluster and for each, all the types name DESCRIBE FUNCTIONS; List all existing keyspaces in the cluster and for each, all the functions name and arguments DESCRIBE AGGREGATES; List all existing keyspaces in the cluster and for each, all the aggregates name and arguments DESCRIBE MATERIALIZED VIEWS; List all existing keyspaces in the cluster and for each, all the materialized view name DESCRIBE KEYSPACE <keyspace name>; Describe the given keyspace configuration and all its table details (name, columns, ...) DESCRIBE TABLE (<keyspace name>).<table name>; Describe the given table. If the keyspace is not provided, the currentlogged in keyspace is used. If there is no logged in keyspace,the default system keyspace is used. If no table is found, an error message is raised DESCRIBE TYPE (<keyspace name>).<type name>; Describe the given type(UDT). If the keyspace is not provided, the currentlogged in keyspace is used. If there is no logged in keyspace,the default system keyspace is used. If no type is found, an error message is raised DESCRIBE FUNCTION (<keyspace name>).<function name>; Describe the given function. If the keyspace is not provided, the currentlogged in keyspace is used. If there is no logged in keyspace,the default system keyspace is used. If no function is found, an error message is raised DESCRIBE AGGREGATE (<keyspace name>).<aggregate name>; Describe the given aggregate. If the keyspace is not provided, the currentlogged in keyspace is used. If there is no logged in keyspace,the default system keyspace is used. If no aggregate is found, an error message is raised DESCRIBE MATERIALIZED VIEW (<keyspace name>).<view name>; Describe the given materialized view. If the keyspace is not provided, the currentlogged in keyspace is used. If there is no logged in keyspace,the default system keyspace is used. If no materialized view is found, an error message is raised Please note that each DESCRIBE command should be ended by a semi-colon.II Schema Display
The schema objects (cluster, keyspace, table, type, view, function & aggregate) are displayed in a tabular format. There is a drop-down menu on the top left corner to expand objects details. On the top right menu is shown the Icon legend.
Sometimes you want to be able to pass runtime query parameters to your statements.Those parameters are not part of the CQL specs and are specific to the interpreter.Below is the list of all parameters:
Query Parameters
Parameter Syntax Description Consistency Level @consistency=value Apply the given consistency level to all queries in the paragraph Serial Consistency Level @serialConsistency=value Apply the given serial consistency level to all queries in the paragraph Timestamp @timestamp=long value Apply the given timestamp to all queries in the paragraph.
Please note that timestamp value passed directly in CQL statement will override this valueRetry Policy @retryPolicy=value Apply the given retry policy to all queries in the paragraph Fetch Size @fetchSize=int value Apply the given fetch size to all queries in the paragraph Request Timeout @requestTimeOut=int value Apply the given request timeout in millisecs to all queries in the paragraph
Some parameters only accept restricted values:Allowed Values
Parameter Possible Values Consistency Level ALL, ANY, ONE, TWO, THREE, QUORUM, LOCAL_ONE, LOCAL_QUORUM, EACH_QUORUM Serial Consistency Level SERIAL, LOCAL_SERIAL Timestamp Any long value Retry Policy DEFAULT, DOWNGRADING_CONSISTENCY, FALLTHROUGH, LOGGING_DEFAULT,LOGGING_DOWNGRADING, LOGGING_FALLTHROUGH Fetch Size Any integer value Request Timeout Any integer value Please note that you should not add semi-colon (;) at the end of each parameter statementSome example:CREATE TABLE IF NOT EXISTS spark_demo.ts(key int PRIMARY KEY,value text);TRUNCATE spark_demo.ts;// Timestamp in the past@timestamp=10// Force timestamp directly in the first insertINSERT INTO spark_demo.ts(key,value) VALUES(1,'first insert') USING TIMESTAMP 100;// Select some data to make the clock turnSELECT * FROM spark_demo.albums LIMIT 100;// Now insert using the timestamp parameter set at the beginning(10)INSERT INTO spark_demo.ts(key,value) VALUES(1,'second insert');// Check for the result. You should see 'first insert'SELECT value FROM spark_demo.ts WHERE key=1;
Some remarks about query parameters:I Syntax
For performance reason, it is better to prepare statements before-hand and reusethem later by providing bound values. This interpreter provides 3 commands to handle prepared andbound statements:
Example:@prepare[statement-name]=...@bind[statement-name]=’text’, 1223, ’2015-07-30 12:00:01’, null, true, [‘list_item1’, ’list_item2’]@bind[statement-name-with-no-bound-value]@remove_prepare[statement-name]
II @prepare
You can use the syntax "@prepare[statement-name]=SELECT ..." to create a prepared statement.The statement-name is mandatory because the interpreter prepares the given statement with theJava driver and saves the generated prepared statement in an internal map, using the providedstatement-name as search key.
Please note that this internal prepared statement map is shared with all notebooksand all paragraphs because there is only one instance of the interpreter for CassandraIf the interpreter encounters many @prepare for the same statement-name (key),only the first statement will be taken into account.
Example:@prepare[select]=SELECT * FROM spark_demo.albums LIMIT ?@prepare[select]=SELECT * FROM spark_demo.artists LIMIT ?
For the above example, the prepared statement is "SELECT * FROM spark_demo.albums LIMIT ?"."SELECT * FROM spark_demo.artists LIMIT ?" is ignored because an entry already exists in theprepared statements map with the key select.
In the context of Zeppelin, a notebook can be scheduled to be executed at regular interval,thus it is necessary to avoid re-preparing many time the same statement (considered an anti-pattern).III @bind
Once the statement is prepared (possibly in a separated notebook/paragraph). You can bind values to it:
@bind[select_first]=10
Bound values are not mandatory for the @bind statement.However if you provide bound values, they need to comply to some syntax:It is possible to use the @bind statement inside a batch:BEGIN BATCH@bind[insert_user]='jdoe','John DOE'UPDATE users SET age = 27 WHERE login='hsue';APPLY BATCH;
IV @remove_prepare
To avoid for a prepared statement to stay forever in the prepared statement map, you can use the @remove_prepare[statement-name] syntaxto remove it. Removing a non-existing prepared statement yields no error.
Instead of hard-coding your CQL queries, it is possible to use Zeppelin dynamic form syntax to inject simple value or multiple choices forms.The legacy mustache syntax ( {{ }} ) to bind input text and select form is still supported but is deprecated and will be removed in future releases.
-- Legacy syntax --
The syntax for simple parameter is: {{input_Label=default value}}.The default value is mandatory because the first time the paragraph is executed,we launch the CQL query before rendering the form so at least one value should be provided.
The syntax for multiple choices parameter is: {{input_Label=value1 | value2 | … | valueN }}.By default the first choice is used for CQL query the first time the paragraph is executed.-- End legacy syntax --
Example:#Secondary index on performer styleSELECT name, country, performerFROM spark_demo.performersWHERE name='${performer=Sheryl Crow|Doof|Fanfarlo|Los Paranoia}'AND styles CONTAINS '${style=Rock}';
In the above example, the first CQL query will be executed for performer='Sheryl Crow'AND style='Rock'. For subsequent queries, you can change the value directly using the form.Please note that we enclosed the {{ }} block between simple quotes (') because Cassandra expects a String here.We could have also use the ${style='Rock'} syntax but this time, the valuedisplayed on the form is 'Rock' and not Rock.It is also possible to use dynamic forms for prepared statements:
@bind[select]=='${performer=Sheryl Crow|Doof|Fanfarlo|Los Paranoia}', '${style=Rock}'The Cassandra interpreter comes with some some configuration values for the Java driver:Interpreter Configuration
Parameter Default Value cassandra.cluster Test Cluster cassandra.compression.protocol NONE, possible values: LZ4, SNAPPY cassandra.credentials.password none cassandra.credentials.username none cassandra.hosts localhost cassandra.interpreter.parallelism 10 cassandra.keyspace system cassandra.load.balancing.policy DEFAULT, or a FQCN of a custom class cassandra.max.schema.agreement.wait.second 10 cassandra.native.port 9042 cassandra.pooling.core.connection.per.host.local Protocol V2 and below: 2, V3 and above: 1 cassandra.pooling.core.connection.per.host.remote Protocol V2 and below: 1, V3 and above: 1 cassandra.pooling.heartbeat.interval.seconds 30 cassandra.pooling.idle.timeout.seconds Test Cluster cassandra.pooling.max.connection.per.host.local Protocol V2 and below: 8, V3 and above: 1 cassandra.pooling.max.connection.per.host.remote Protocol V2 and below: 2, V3 and above: 1 cassandra.pooling.max.request.per.connection.local Protocol V2 and below: 128, V3 and above: 1024 cassandra.pooling.max.request.per.connection.remote Protocol V2 and below: 128, V3 and above: 256 cassandra.pooling.new.connection.threshold.local Protocol V2 and below: 100, V3 and above: 800 cassandra.pooling.new.connection.threshold.remote Protocol V2 and below: 100, V3 and above: 200 cassandra.pooling.pool.timeout.millisecs 5000 cassandra.protocol.version 4 cassandra.query.default.consistency ONE cassandra.query.default.fetchSize 5000 cassandra.query.default.serial.consistency SERIAL cassandra.reconnection.policy DEFAULT, or a FQCN of a custom class cassandra.retry.policy DEFAULT, or a FQCN of a custom class cassandra.socket.connection.timeout.millisecs 500 cassandra.socket.read.timeout.millisecs 12000 cassandra.socket.tcp.no_delay true cassandra.speculative.execution.policy DEFAULT, or a FQCN of a custom class It is possible to execute many paragraphs in parallel. However, at the back-end side, we’re still using synchronous queries. Asynchronous execution is only possible when it is possible to return a Future value in the InterpreterResult. It may be an interesting proposal for the Zeppelin project.
Recently, Zeppelin allows you to choose the level of isolation for your interpreters (seeInterpreter Binding Mode ).
Long story short, you have 3 available bindings:
Using the shared binding, the samecom.datastax.driver.core.Sessionobject is used for all notes and paragraphs.Consequently, if you use the USE keyspace name; statement to log into a keyspace,it will change the keyspace for all current users of the Cassandra interpreter because we only create 1com.datastax.driver.core.Sessionobject per instance of Cassandra interpreter.
The same remark does apply to the prepared statement hash map, it is shared by all users using the same instance of Cassandra interpreter.
When using scoped binding, in the same JVMZeppelin will create multiple instances of the Cassandra interpreter,thus multiplecom.datastax.driver.core.Sessionobjects.Beware of resource and memory usage using this binding !
The isolated mode is the most extreme and will create as many JVM/com.datastax.driver.core.Sessionobject as there are distinct notes.3.0 :If you encounter a bug for this interpreter, please create a JIRA ticket and ping me on Twitter at @doanduyhaizeppelin.interpreters -org.apache.zeppelin.spark.SparkInterpreter,org.apache.zeppelin.spark.PySparkInterpreter,org.apache.zeppelin.rinterpreter.RRepl,org.apache.zeppelin.rinterpreter.KnitR,org.apache.zeppelin.spark.SparkRInterpreter,org.apache.zeppelin.spark.SparkSqlInterpreter,org.apache.zeppelin.spark.DepInterpreter,org.apache.zeppelin.markdown.Markdown,org.apache.zeppelin.angular.AngularInterpreter,org.apache.zeppelin.shell.ShellInterpreter,org.apache.zeppelin.hive.HiveInterpreter,org.apache.zeppelin.tajo.TajoInterpreter,org.apache.zeppelin.file.HDFSFileInterpreter,org.apache.zeppelin.flink.FlinkInterpreter,,org.apache.zeppelin.python.PythonInterpreter,org.apache.zeppelin.lens.LensInterpreter,org.apache.zeppelin.ignite.IgniteInterpreter,org.apache.zeppelin.ignite.IgniteSqlInterpreter,org.apache.zeppelin.cassandra.CassandraInterpreter,org.apache.zeppelin.geode.GeodeOqlInterpreter,org.apache.zeppelin.postgresql.PostgreSqlInterpreter,org.apache.zeppelin.jdbc.JDBCInterpreter,org.apache.zeppelin.phoenix.PhoenixInterpreter,org.apache.zeppelin.kylin.KylinInterpreter,org.apache.zeppelin.elasticsearch.ElasticsearchInterpreter,org.apache.zeppelin.scalding.ScaldingInterpreter,org.apache.zeppelin.alluxio.AlluxioInterpreter,org.apache.zeppelin.hbase.HbaseInterpreter,org.apache.zeppelin.livy.LivySparkInterpreter,org.apache.zeppelin.livy.LivyPySparkInterpreter,org.apache.zeppelin.livy.LivySparkRInterpreter,org.apache.zeppelin.livy.LivySparkSQLInterpreter +org.apache.zeppelin.spark.SparkInterpreter,org.apache.zeppelin.spark.PySparkInterpreter,org.apache.zeppelin.rinterpreter.RRepl,org.apache.zeppelin.rinterpreter.KnitR,org.apache.zeppelin.spark.SparkRInterpreter,org.apache.zeppelin.spark.SparkSqlInterpreter,org.apache.zeppelin.spark.DepInterpreter,org.apache.zeppelin.markdown.Markdown,org.apache.zeppelin.angular.AngularInterpreter,org.apache.zeppelin.shell.ShellInterpreter,org.apache.zeppelin.file.HDFSFileInterpreter,org.apache.zeppelin.flink.FlinkInterpreter,,org.apache.zeppelin.python.PythonInterpreter,org.apache.zeppelin.lens.LensInterpreter,org.apache.zeppelin.ignite.IgniteInterpreter,org.apache.zeppelin.ignite.IgniteSqlInterpreter,org.apache.zeppelin.cassandra.CassandraInterpreter,org.apache.zeppelin.geode.GeodeOqlInterpreter,org.apache.zeppelin.postgresql.PostgreSqlInterpreter,org.apache.zeppelin.jdbc.JDBCInterpreter,org.apache.zeppelin.phoenix.PhoenixInterpreter,org.apache.zeppelin.kylin.KylinInterpreter,org.apache.zeppelin.elasticsearch.ElasticsearchInterpreter,org.apache.zeppelin.scalding.ScaldingInterpreter,org.apache.zeppelin.alluxio.AlluxioInterpreter,org.apache.zeppelin.hbase.HbaseInterpreter,org.apache.zeppelin.livy.LivySparkInterpreter,org.apache.zeppelin.livy.LivyPySparkInterpreter,org.apache.zeppelin.livy.LivySparkRInterpreter,org.apache.zeppelin.livy.LivySparkSQLInterpreter Comma separated interpreter configurations. First interpreter become a default 
Zeppelin
- (0.6.0-SNAPSHOT)
+ {{site.ZEPPELIN_VERSION}}