无法使用 spark (java) 从 Cassandra 获取数据答案

【问题标题】：Unable to fetch data from Cassandra using spark (java)无法使用 spark (java) 从 Cassandra 获取数据
【发布时间】：2017-06-15 15:28:12
【问题描述】：

我是 Cassandra 和 Spark 的新手，我正在尝试使用 spark 从数据库中获取数据。我为此目的使用Java。问题是没有抛出异常或发生错误，但我仍然无法获取数据。在下面找到我的代码 -

    SparkConf sparkConf = new SparkConf();
    sparkConf.setAppName("Spark-Cassandra Integration");
    sparkConf.setMaster("local[4]");
    sparkConf.set("spark.cassandra.connection.host", "stagingHost22");
    sparkConf.set("spark.cassandra.connection.port", "9042");

    sparkConf.set("spark.cassandra.connection.timeout_ms", "5000");
    sparkConf.set("spark.cassandra.read.timeout_ms", "200000");


    JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);
    String keySpaceName = "testKeySpace";
    String tableName = "testTable";

    CassandraJavaRDD<CassandraRow> cassandraRDD = CassandraJavaUtil.javaFunctions(javaSparkContext).cassandraTable(keySpaceName, tableName);

    final ArrayList dataList = new ArrayList();
    JavaRDD<String> userRDD = cassandraRDD.map(new Function<CassandraRow, String>() {

        private static final long serialVersionUID = -165799649937652815L;


        public String call(CassandraRow row) throws Exception {
            System.out.println("Inside RDD call");
            dataList.add(row);
            return "test";
        }
    });
    System.out.println( "data Size -" + dataList.size());

Cassandra 和 spark maven 依赖项是 -

      <dependency>
        <groupId>com.datastax.cassandra</groupId>
        <artifactId>cassandra-driver-core</artifactId>
        <version>3.0.0</version>
    </dependency>

    <dependency>
        <groupId>com.datastax.cassandra</groupId>
        <artifactId>cassandra-driver-mapping</artifactId>
        <version>3.0.0</version>
    </dependency>

    <dependency>
        <groupId>com.datastax.cassandra</groupId>
        <artifactId>cassandra-driver-extras</artifactId>
        <version>3.0.0</version>
    </dependency>
    <dependency>
        <groupId>com.sparkjava</groupId>
        <artifactId>spark-core</artifactId>
        <version>2.5.4</version>
    </dependency>
     <dependency>
        <groupId>com.datastax.spark</groupId>
        <artifactId>spark-cassandra-connector_2.10</artifactId>
        <version>2.0.0-M3</version>
    </dependency>
     <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.10</artifactId>
        <version>1.4.0</version>
    </dependency>

    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.10</artifactId>
        <version>1.3.0</version>
    </dependency>

这可以确定 stagingHost22 主机具有密钥空间 - testKeySpace 和表名 - testTable 的 cassandra 数据。查找下面的查询输出 -

cqlsh:testKeySpace> select count(*) from testTable;

计数

34

（1 行）

有人可以建议我在这里缺少什么吗？

提前致谢。

热烈的问候，

Vibhav

【问题讨论】：

标签： java maven apache-spark cassandra spark-cassandra-connector

【解决方案1】：

您当前的代码没有执行任何 Spark 操作。因此没有加载数据。

请参阅 Spark 文档以了解 Spark 中的转换和操作之间的区别： http://spark.apache.org/docs/latest/programming-guide.html#rdd-operations

此外，在使用 Cassandra 连接器时，通常不需要将 CassandraRows 添加到 ArrayList。我建议先实现一个简单的选择（遵循 Spark-Cassandra-Connector 文档）。如果这有效，您可以根据需要扩展此代码。

查看以下链接，了解如何使用连接器加载数据的示例：

【讨论】：