如何使用 Java API 将 DataSet<Row> ds 插入 cassandra答案

【问题标题】：How to insert DataSet<Row> ds into cassandra with Java API如何使用 Java API 将 DataSet<Row> ds 插入 cassandra
【发布时间】：2018-02-16 18:53:31
【问题描述】：

Spark Cassandra 连接器 2.11-2.0.5 所需的示例代码，无法将数据集直接插入 cassandra db

【问题讨论】：

您需要更具体一点。您到底尝试了哪些方法，收到了哪些错误消息？

标签： apache-spark-sql spark-dataframe spark-cassandra-connector cassandra-3.0

【解决方案1】：

添加以下 Cassandra 依赖项

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

将您的 Cassandra 数据库配置详细信息设置为 Spark

SparkConf sparkConf = new SparkConf();
sparkConf.setAppName("Spark-Cassandra Integration");
sparkConf.setMaster("local[4]");
sparkConf.set("spark.cassandra.connection.host", "127.0.0.1");
sparkConf.set("spark.cassandra.connection.native.port", "9042");
sparkConf.set("spark.cassandra.connection.rpc.port", "9160");
sparkConf.set("spark.cassandra.connection.timeout_ms", "5000");
sparkConf.set("spark.cassandra.read.timeout_ms", "200000");
sparkConf.set("spark.cassandra.auth.username", "test_user");
sparkConf.set("spark.cassandra.auth.password", "test_password");

创建 Spark 上下文

JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);

获得 RDD 后，通过提供键空间名称和表名将其保存到 Cassandra 数据库

CassandraJavaUtil.javaFunctions(userRDD).writerBuilder(keySpaceName, tableName, CassandraJavaUtil.mapToRow(UserBean.class)).saveToCassandra();

更多详情可以查看我的博文Save data to Cassandra tables using Apache Spark

【讨论】：