使用java语言的spark cassandra连接应该使用哪些版本?

kyks70gy  于 2021-06-10  发布在  Cassandra
关注(0)|答案(1)|浏览(278)

我试图用java语言连接spark和cassandra数据库。连接Spark和Cassandra我使用的是最新版本的SparkCassandra连接器即2.4.0。目前我可以连接Spark和Cassandra使用连接器。我正在获取rdd格式的数据,但我无法从该数据结构中读取数据。如果我使用行读取器工厂作为cassandratable()的第三个参数

> Wrong 3rd argument type. Found:
> 'java.lang.Class<com.journaldev.sparkdemo.JohnnyDeppDetails>',
> required:
> 'com.datastax.spark.connector.rdd.reader.RowReaderFactory<T>'

有人能告诉我应该用哪个版本吗?这里有什么问题吗?
cassandratablescanjavardd pricesrd2=cassandrajavautil.javafunctions(sc).cassandratable(keyspace,table,johnnyDepDetails.class);
我的pom.xml:

<!-- Import Spark -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
        <version>2.4.0</version>
    </dependency>

    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.11</version>
        <scope>test</scope>
    </dependency>

    <!-- https://mvnrepository.com/artifact/com.datastax.spark/spark-cassandra-connector -->
    <dependency>
        <groupId>com.datastax.spark</groupId>
        <artifactId>spark-cassandra-connector_2.11</artifactId>
        <version>2.4.0</version>
    </dependency>

    <dependency>
        <groupId>com.datastax.spark</groupId>
        <artifactId>spark-cassandra-connector-java_2.10</artifactId>
        <version>1.5.0-M2</version>
    </dependency>

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

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

    <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.12</artifactId>
        <version>2.4.0</version>
    </dependency>

</dependencies>
pbwdgjma

pbwdgjma1#

您需要创建一个 RowReaderFactory 使用 mapRowTo 函数,如下所示(这是我的示例):

CassandraJavaRDD<UUIDData> uuids = javaFunctions(spark.sparkContext())
      .cassandraTable("test", "utest", mapRowTo(UUIDData.class));

在写回时,可以通过将类转换为相应的工厂 mapToRow 功能。

相关问题