带有 java 源代码的 Spark-Cassandra Maven 项目进行 scala-lib 调用答案

【问题标题】：Spark-Cassandra Maven project with java source making scala-lib calls带有 java 源代码的 Spark-Cassandra Maven 项目进行 scala-lib 调用
【发布时间】：2015-07-17 00:30:48
【问题描述】：

我对 Spark 和 Cassandra 都很陌生，需要一些指导。我正在建立一个使用 Spark v1.3.1 和 Cassandra v2.0.14 的 maven 项目。我正在尝试以下操作：

1) 使用以下方法与Oracle DB建立连接进行数据输入；利用 Spark 1.3.0 的新数据帧：http://www.sparkexpert.com/2015/03/28/loading-database-data-into-spark-using-data-sources-api/

2) 使用 spark-cassandra-connector 进行后者之间的连接；在github上找到。

3) 一旦我在 DataFrame 中有数据库数据，我应该能够转换为 JavaRDD 类型并推送到 Cassandra 键空间，如下所示：http://www.datastax.com/dev/blog/accessing-cassandra-from-spark-in-java

4) 简而言之：[Oracle DB][Cassandra]

我遇到的问题是在我的 Java 代码中的 Scala-lib 调用期间（上面的第 1 步）；更具体地说，在加载函数调用期间：DataFrame jdbcDF = sqlContext.load(“jdbc”, options);

运行时错误： java.lang.ClassNotFoundException: scala.collection.GenTraversableOnce$class”

尽管在我的 pom.xml 文件中尝试了推荐的 2.10.X Scala 的几个不同版本，但仍出现上述错误。根据我之前的研究，我认为这可能是 Spark-Scala 兼容性问题。我还读到我需要在我的类路径中包含 scala-lib.jar，但我不确定如何使用 maven 执行此操作。对此有任何想法吗？我在下面包含了 pom.xml 和 java 代码：

POM.XML

<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>
    <groupId>com.dev</groupId>
    <artifactId>spark-cassandra</artifactId>
    <version>0.0.1-SPARK-CASSANDRA</version>

    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>1.3.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
            <version>1.3.1</version>
        </dependency>

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.35</version>
        </dependency>
	<dependency>
	    <groupId>com.oracle</groupId>
	    <artifactId>ojdbc6</artifactId>
	    <version>11.2.0</version>
	</dependency>
	
	
	<dependency>
	    <groupId>com.datastax.spark</groupId>
	    <artifactId>spark-cassandra-connector_2.10</artifactId>
	    <version>1.0.0-rc4</version>
	</dependency>
	<dependency>
	    <groupId>com.datastax.spark</groupId>
	    <artifactId>spark-cassandra-connector-java_2.10</artifactId>
	    <version>1.0.0-rc4</version>
	</dependency>
	<dependency>
	    <groupId>com.datastax.cassandra</groupId>
	    <artifactId>cassandra-driver-core</artifactId>
	    <version>2.1.5</version>
	</dependency>
	
	<dependency>
	    <groupId>org.apache.spark</groupId>
	    <artifactId>spark-streaming_2.10</artifactId>
	    <version>1.3.1</version>
	</dependency>
	<dependency>
	    <groupId>com.dev.cassandra</groupId>
	    <artifactId>spark-cassandra</artifactId>
	    <version>1.0</version>
	</dependency>
	
	<dependency>	
	    <groupId>org.scala-lang</groupId>
	    <artifactId>scala-library</artifactId>
	    <version>2.10.3</version>
	</dependency>
	
	<dependency>
	<groupId>org.scala-lang</groupId>
	<artifactId>scala-compiler</artifactId>
	<version>2.10.3</version>
	</dependency>
	<!--
	<dependency>
	<groupId>org.scala-lang</groupId>
	<artifactId>scala-reflect</artifactId>
	<version>2.10.0-M1</version>
	</dependency>
	-->

		
	<!--
	
	<dependency>
	    <groupId>org.scala-lang</groupId>
	    <artifactId>scala-swing</artifactId>
	    <version>2.10.0-M1</version>
	</dependency>
	-->
	
    </dependencies>
   

    <build>
        <pluginManagement>
	  
            <plugins>
                <plugin>
                    <groupId>net.alchim31.maven</groupId>
                    <artifactId>scala-maven-plugin</artifactId>
                    <version>3.1.5</version>
                </plugin>
		<plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.3</version>
                <configuration>
                    <source>1.7</source>
                    <target>1.7</target>
		    <mainClass>com.dev.cassandra.Main</mainClass>
		    <cleanupDaemonThreads>false</cleanupDaemonThreads>
                    <compilerArgument>-Xlint:all</compilerArgument>
                    <showWarnings>true</showWarnings>
                    <showDeprecation>true</showDeprecation>
                </configuration>
	      </plugin>
            </plugins>
        </pluginManagement>

        <plugins>

            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <executions>
                    <execution>
                        <id>scala-compile-first</id>
                        <phase>process-resources</phase>
                        <goals>
                            <goal>add-source</goal>
                            <goal>compile</goal>
                        </goals>
                    </execution>
                    <execution>
                        <id>scala-test-compile</id>
                        <phase>process-test-resources</phase>
                        <goals>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>

            <!-- Plugin to create a single jar that includes all dependencies 
            <plugin>
		<groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>2.4</version>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                    <archive>
                        <manifest>
                            <mainClass>com.dev.cassandra.Main</mainClass>
                        </manifest>
                    </archive>
                </configuration>
                <executions>
                    <execution>
			<id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
		-->
        </plugins>
    </build>				

</project>

JAVA 代码：

package com.dev.cassandra;

import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import java.sql.*;

import org.apache.spark.*;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.*;	
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import oracle.jdbc.*;

import com.datastax.spark.connector.cql.CassandraConnector;
import static com.datastax.spark.connector.CassandraJavaUtil.*;

public class Main implements Serializable {

    private static final org.apache.log4j.Logger LOGGER = org.apache.log4j.Logger.getLogger(Main.class);

    private static final String JDBC_DRIVER = "oracle.jdbc.driver.OracleDriver";
    private static final String JDBC_USERNAME = "XXXXXO01";
    private static final String JDBC_PWD = "XXXXXO01";
    private static final String JDBC_CONNECTION_URL =
            "jdbc:oracle:thin:" + JDBC_USERNAME + "/" + JDBC_PWD + "@CONNECTION VALUES";

    private transient SparkConf conf;
  
    private Main(SparkConf conf) {
        this.conf = conf;
    }
  
    private void run() {
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(sc);
        generateData(sc);
        compute(sc);
        showResults(sc);
        sc.stop();
    }
  
    private void generateData(JavaSparkContext sc) {
    
      SQLContext sqlContext = new org.apache.spark.sql.SQLContext(sc);
      System.out.println("AFTER SQL CONTEXT");
      
      //Data source options
      Map<String, String> options = new HashMap<>();
      options.put("driver", JDBC_DRIVER);
      options.put("url", JDBC_CONNECTION_URL);
      options.put("dbtable","(SELECT * FROM XXX_SAMPLE_TABLE WHERE ROWNUM <=5)");
      
      CassandraConnector connector = CassandraConnector.apply(sc.getConf());
      
      try{	
	Class.forName(JDBC_DRIVER);
	 
	System.out.println("BEFORE jdbcDF"); 
	 
        //Load JDBC query result as DataFrame
        DataFrame jdbcDF = sqlContext.load("jdbc", options);
        System.out.println("AFTER jdbcDF");

        List<Row> tableRows = jdbcDF.collectAsList();
        
        System.out.println("AFTER tableRows");

        for (Row tableRow : tableRows) {
	    System.out.println();
            LOGGER.info(tableRow);
            System.out.println();
        }
        
	}catch(Exception e){
	  //Handle errors for Class.forName
	  e.printStackTrace();
	}
    }
  
    private void compute(JavaSparkContext sc) {
    }
  
    private void showResults(JavaSparkContext sc) {
    }
    
    
    public static void main(String[] args) throws InterruptedException 
    {
	
      if (args.length != 2) {
            System.err.println("Syntax: com.datastax.spark.dev.cassandra <Spark Master URL> <Cassandra contact point>");
            System.exit(1);
        }
  
	//JavaSparkContext sc = new JavaSparkContext(new SparkConf().setAppName("SparkJdbcDs").setMaster("local[*]"));
	SparkConf conf = new SparkConf().setAppName("SparkJdbcDs").setMaster("local[*]");

	
	//SparkConf conf = new SparkConf();
        //conf.setAppName("SparkJdbcDs");
        //conf.setMaster(args[0]);
        //conf.set("spark.cassandra.connection.host", args[1]);
  
        Main app = new Main(conf);
        app.run();
	
    }
}

提前致谢！

【问题讨论】：

Spark-Cassandra 连接器还不支持 Spark 1.3.x - 请参阅兼容性表：github.com/datastax/spark-cassandra-connector
此外，您的 pom.xml 正在请求 Spark JAR 的 Scala 2.11 版本和 Cassandra JAR 的 Scala 2.10 版本。（基于 Scala 工件 ID 的命名约定，它以您希望为其构建的 Scala 版本结尾。）这些需要（a）彼此一致，（b）与您实际使用的 Scala 版本一致。
@SpiroMichaylov - 非常感谢您的推荐；我的错误，就像你提到的那样，是我在 pom.xml 中调用了错误的 Scala 版本（2.11）。在我的 Spark 依赖项中更改为 2.10 版后，它起作用了。谢谢！
@maasg - 由于 spark-cassandra-connector 不支持 Spark 1.3.x，但我可能无法使用下面帖子中使用的 DataFrame 概念来发送/转换 DataFrame卡桑德拉正确吗？ sparkexpert.com/2015/03/28/…
不，唉，你将无法使用 DataFrame，因为它们是在 1.3.0 中引入的。此外，Cassandra 连接器对 Spark SQL 的支持通常早于 Spark 1.2.0 中引入的外部数据源 API，因此即使对 Spark 1.3.0 的支持出现，它也不一定会立即支持博客文章中描述的方法你参考。（但是，您可以使用Spark SQL support 获得类似的功能——搜索“SQL”。）

标签： java scala maven cassandra apache-spark

【解决方案1】：

pom.xml 请求某些 Spark JAR 的 Scala 2.11 版本

<artifactId>spark-core_2.11</artifactId>

<artifactId>spark-sql_2.11</artifactId>

以及另一个 Spark JAR 和 Cassandra 连接器 JAR 的 Scala 2.10 版本。

<artifactId>spark-streaming_2.10</artifactId>

<artifactId>spark-cassandra-connector_2.10</artifactId>

<artifactId>spark-cassandra-connector-java_2.10</artifactId>

（基于 Scala 工件 ID 的命名约定，以您希望为其构建的 Scala 版本结尾。）

这些需要保持 (a) 彼此一致，并且 (b) 与您实际使用的 Scala 版本保持一致。

【讨论】：

很好，非常感谢