{"id":2325,"date":"2016-08-08T08:06:24","date_gmt":"2016-08-08T08:06:24","guid":{"rendered":"http:\/\/www.doanduyhai.com\/blog\/?p=2325"},"modified":"2018-07-28T07:35:56","modified_gmt":"2018-07-28T07:35:56","slug":"zeppelinsparkcassandra-integration-tutorial","status":"publish","type":"post","link":"https:\/\/www.doanduyhai.com\/blog\/?p=2325","title":{"rendered":"Zeppelin\/Spark\/Cassandra integration tutorial"},"content":{"rendered":"

In this post, I’ll cover in detail all the steps necessary to integrate Apache Zeppelin<\/a><\/strong>, Apache Spark<\/a><\/strong> and Apache Cassandra<\/a><\/strong>.<\/p>\n

For the remaining of this post, Zeppelin == Apache Zeppelin\u2122, Spark == Apache Spark\u2122 and Cassandra == Apache Cassandra\u2122<\/p><\/blockquote>\n

If you are not familiar with Zeppelin, I recommend reading my introduction slides here<\/a><\/strong><\/p>\n

The integration between Spark<\/strong> and Cassandra<\/strong> is achieved using the Spark-Cassandra connector<\/a><\/strong>. <\/p>\n

Natively Zeppelin<\/strong> does support Spark<\/strong> out of the box. But making Zeppelin<\/strong> supporting the Spark-Cassandra integration requires some extra work.<\/p>\n

Zeppelin – Spark workflow<\/h1>\n

With Zeppelin, any interpreter is executed in a separated JVM and it does apply to the Spark interpreter too.<\/p>\n

The interpreter is first launched in the class RemoteInterpreterProcess<\/strong><\/code>:<\/p>\n

\r\n public int reference(InterpreterGroup interpreterGroup) {\r\n        ...\r\n        if (!isInterpreterAlreadyExecuting) {\r\n          try {\r\n            port = RemoteInterpreterUtils.findRandomAvailablePortOnAllLocalInterfaces();\r\n          } catch (IOException e1) {\r\n            throw new InterpreterException(e1);\r\n          }\r\n          CommandLine cmdLine = CommandLine.parse(interpreterRunner);\r\n          cmdLine.addArgument("-d", false);\r\n          cmdLine.addArgument(interpreterDir, false);\r\n          cmdLine.addArgument("-p", false);\r\n          cmdLine.addArgument(Integer.toString(port), false);\r\n          cmdLine.addArgument("-l", false);\r\n          cmdLine.addArgument(localRepoDir, false);\r\n\r\n          executor = new DefaultExecutor();\r\n\r\n          watchdog = new ExecuteWatchdog(ExecuteWatchdog.INFINITE_TIMEOUT);\r\n          executor.setWatchdog(watchdog);\r\n\r\n          running = true;\r\n          ...\r\n<\/pre>\n
 Indeed each interpreter is bootstrapped using the interpreterRunner<\/strong><\/code> which is the shell script $ZEPPELIN_HOME\/bin\/interpreter.sh<\/strong><\/code><\/p>\n
 Depending on the interpreter type and run mode, the execution is launched with a different set of environment. Extract of the $ZEPPELIN_HOME\/bin\/interpreter.sh<\/strong><\/code> script:<\/p>\n
\r\nif [[ -n "${SPARK_SUBMIT}" ]]; then\r\n    ${SPARK_SUBMIT} --class ${ZEPPELIN_SERVER} --driver-class-path "${ZEPPELIN_CLASSPATH_OVERRIDES}:${CLASSPATH}" --driver-java-options "${JAVA_INTP_OPTS}" ${SPARK_SUBMIT_OPTIONS} ${SPARK_APP_JAR} ${PORT} &\r\nelse\r\n    ${ZEPPELIN_RUNNER} ${JAVA_INTP_OPTS} ${ZEPPELIN_INTP_MEM} -cp ${ZEPPELIN_CLASSPATH_OVERRIDES}:${CLASSPATH} ${ZEPPELIN_SERVER} ${PORT} &\r\nfi\r\n<\/pre>\n
There is a small detail here that is critical for the integration of the Spark-Cassandra connector, which is the classpath<\/strong> used to launch the interpreter process. The idea is to include the Spark-Cassandra connector dependencies in this classpath so that Zeppelin<\/strong> can access Cassandra<\/strong> using Spark<\/strong><\/p>\n
Configuration matrix<\/h1>\n
 There are many parameters and configurations to run Zeppelin<\/strong> with Spark<\/strong> and Cassandra<\/strong>:<\/p>\n
    \n
  1. Standard Zeppelin binaries<\/li>\n
  2. Custom Zeppelin build with the Spark-Cassandra connector<\/li>\n
  3. Zeppelin connecting to the local Spark runner<\/li>\n
  4. Zeppelin connecting to a stand-alone Spark cluster<\/li>\n
  5. Using Zeppelin with OSS Spark<\/li>\n
  6. Using Zeppelin with DSE (Datastax Enterprise<\/a><\/strong>)<\/li>\n<\/ol>\n
    Standard Zeppelin build with local Spark<\/h1>\n
    If you are using the default Zeppelin<\/strong> binaries (downloaded from the official repo), to make the Spark-Cassandra integration work, you would have to<\/p>\n
      \n
    1. In the Interpreter<\/strong> menu, add the property spark.cassandra.connection.host<\/em> to the Spark interpreter. The value should point to a single or a list of IP addresses of your Cassandra cluster
      \n  
      \"spark_cassandra_connection_host\"
      spark_cassandra_connection_host<\/p><\/div>\n <\/li>\n
    2. last but not least, you’d have to add also the Spark-Cassandra connector as dependency to the interpreter
      \n 
      \"spark_cassandra_dependencies\"
      spark_cassandra_dependencies<\/p><\/div><\/p>\n
        when adding the dependency and the property, do not forget to click on the + icon<\/strong> to force Zeppelin to add your change otherwise it will be lost<\/p><\/blockquote>\n<\/li>\n<\/ol>\n
       What happens at runtime is Zeppelin will download the declared dependencie(s) and all its transitive dependencie(s) from Maven central and\/or from your local Maven repository (if any).<\/p>\n
       Those dependencies will then be stored inside the local repository folder defined by the property: zeppelin.dep.localrepo<\/em>.<\/p>\n
       Also, if you go back to the interpreter configuration menu (after a successful run), you’ll see a new property added by Zeppelin: zeppelin.interpreter.localRepo<\/strong><\/code><\/p>\n
      \"interpreter_repo\"
      interpreter_repo<\/p><\/div>\n
       The last string in the folder (2BTPVTBVH<\/strong> in the example) is the id of the interpreter instance. All transitive dependencies are downloaded and stored as jar files inside $ZEPPELIN_HOME\/local-repo\/<INTERPRETER_ID><\/em><\/code> and their content (.class files) is extracted into $ZEPPELIN_HOME\/local-repo<\/em><\/code><\/p>\n
      If your Zeppelin server is behind corporate firewall<\/strong>, the download will fail so Spark won’t be able to connect to Cassandra (you’ll get a ClassNotFoundException<\/code> in the Spark interpreter logs).<\/strong><\/p>\n
       The solution in this case is:<\/p>\n
        \n
      1. either download manually all the dependencies and put them into the folder zeppelin.dep.localrepo<\/em><\/li>\n
      2. or build Zeppelin with the Spark-Cassandra connector integrated (see right after)<\/li>\n<\/ol>\n
        Custom Zeppelin build with local Spark<\/h1>\n
         You’ll need to build Zeppelin yourself, using one of the available Maven profiles cassandra-spark-1.x<\/strong> to get the correct Spark version.<\/p>\n
         Those profiles are defined in the $ZEPPELIN_HOME\/spark-dependencies\/pom.xml<\/strong><\/code> file.<\/p>\n
         For each cassandra-spark-1.x<\/strong>, you can override the defined Spark version using the -Dspark.version=x.y.z<\/em><\/code> flag for the build. To change the Spark-Cassandra connector version, you’ll need to edit the $ZEPPELIN_HOME\/spark-dependencies\/pom.xml<\/strong><\/code> file yourself. Similarly if you want to use the latest version of the Spark-Cassandra connector and a profile does not exist, just edit the file and add your own profile.<\/p>\n
         In a nutshell, the build command is<\/p>\n
        \r\n>mvn clean package -Pbuild-distr -Pcassandra-spark-1.x -DskipTests \r\n<\/pre>\n
        or with the Spark version manually forced:<\/p>\n
        \r\n>mvn clean package -Pbuild-distr -Pcassandra-spark-1.x -Dspark.version=x.y.z -DskipTests \r\n<\/pre>\n
         This will force Zeppelin to add all transitive dependencies for the Spark-Cassandra connector into the big\/fat jar file located in $ZEPPELIN_HOME\/interpreter\/spark\/dep\/zeppelin-spark-dependencies-<ZEPPELIN_VERSION>.jar<\/em><\/code><\/p>\n
         One easy way to verify that the Spark-Cassandra connector has been correctly embedded into this file is to copy it somewhere and extract its content to check using the command jar -xvf zeppelin-spark-dependencies-<ZEPPELIN_VERSION>.jar<\/em><\/code> <\/p>\n
         Once built, you can use this special version of Zeppelin without declaring any dependency to the Spark-Cassandra connector. You still have to set the spark.cassandra.connection.host<\/strong><\/code> property on the Spark interpreter<\/p>\n
        Zeppelin connecting to a stand-alone OSS Spark cluster<\/h1>\n
         Until now, we have supposed that you are using the local Spark mode of Zeppelin (master = local[*]<\/em><\/strong>). In this section, we want Zeppelin to connect to an existing stand-alone Spark cluster (Spark running on Yarn and Mesos is not covered here because it is recommended to run Spark in stand-alone mode with Cassandra to benefit from data-locality<\/strong>).<\/p>\n
         First, you’ll need to set the Spark master property for the Spark interpreter. Instead of local[*]<\/strong>, put a real address like spark:\/\/x.y.z:7077<\/strong>.<\/p>\n
        \"spark_master_url\"
        spark_master_url<\/p><\/div>\n
         The extract of the shell script from the first section showed that Zeppelin will invoke the spark-submit<\/strong><\/code> command, passing its own Spark jar with all the transitive dependencies using the parameter --driver-class-path<\/em><\/code>. <\/p>\n
         But where does Zeppelin fetches all the dependencies jar ? From the local repository seen earlier !!!<\/p>\n
        As a consequence, if you add the Spark-Cassandra connector as dependency (standard Zeppelin build) and you run against a stand-alone Spark cluster, it will fail because the local repository will be empty!!!. Run first a simple Spark job in local Spark mode to let Zeppelin a chance to download the dependencies before switching to the stand-alone Spark<\/strong><\/p><\/blockquote>\n
         But it’s not sufficient, on your stand-alone Spark cluster, you must also add the Spark-Cassandra connector dependencies into Spark classpath so that the workers can connect to Cassandra.<\/p>\n
         How do to that ?<\/p>\n
          \n
        1. edit $SPARK_HOME\/conf\/spark-env.sh<\/em><\/code> file and add the Spark-Cassandra dependencies to the SPARK_CLASSPATH<\/strong> variable.
          \n  
          \"spark_classpath\"
          spark_classpath<\/p><\/div><\/p>\n
            As you can see, it’s not just the simple spark-cassandra connector jar we need but the assembly jar<\/strong> e.g. the fat jar which includes all transitive dependencies<\/strong>.<\/p>\n
            To get this jar, you’ll have to build it yourself:<\/p>\n