Datastax spark cassandra 连接器 - 将 DF 写入 cassandra 表答案

【问题标题】：Datastax spark cassandra connector - writing DF to cassandra tableDatastax spark cassandra 连接器 - 将 DF 写入 cassandra 表
【发布时间】：2017-03-04 09:42:36
【问题描述】：

我们最近开始使用 Scala、Spark 和 Cassandra 进行大数据项目，我对所有这些技术都很陌生。我正在尝试对 cassandra 表进行简单的读写任务。如果我将属性名和列名全部保留为小写或蛇形（unserscores），我可以实现这一点，但我想在我的 scala 代码中使用驼峰式大小写。有没有更好的方法可以在 Scala 中使用驼峰格式，在 cassandra 中使用蛇形格式。

我们正在使用

scala - 2.10.5 spark - 1.6.2 datastax spark-cassandra-connector - 1.6.0 cassandra - 3.0.9.1346 数据税企业 - 5.0.3

卡桑德拉表

CREATE TABLE dev.castable (
id int PRIMARY KEY,
long_name text,
name text,
short_name text)

Scala 代码

    val conf = new SparkConf()
        .setAppName("TestHelper")
        .setMaster("local")
        .set("spark.cassandra.connection.host","127.0.01")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)

    println("writing data to cassandra")
    val df = sqlContext.createDataFrame(List(new MyRow(2,Option("long name"), "ss", Option("short name"))))
    df.write //*** this is not working
      .cassandraFormat("castable", "dev")
      .mode(SaveMode.Append)
      .save()

    println("reading data from cassandra") //*** This is working fine
    val rdd = sc.cassandraTable[MyRow]("dev", "castable")
    rdd.foreach(println)

例外

Exception in thread "main" java.util.NoSuchElementException: Columns not found in table dev.castable: longName, shortName
at com.datastax.spark.connector.SomeColumns.selectFrom(ColumnSelector.scala:38)
at com.datastax.spark.connector.writer.TableWriter$.apply(TableWriter.scala:268)
at com.datastax.spark.connector.RDDFunctions.saveToCassandra(RDDFunctions.scala:36)
at org.apache.spark.sql.cassandra.CassandraSourceRelation.insert(CassandraSourceRelation.scala:67)
at org.apache.spark.sql.cassandra.DefaultSource.createRelation(DefaultSource.scala:85)
at org.apache.spark.sql.execution.datasources.ResolvedDataSource$.apply(ResolvedDataSource.scala:222)
at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:148)
at com.aktana.spark.util.LocalTestDriver$.main(LocalTestDriver.scala:38)

我读到 spark-cassandra-connector 会自动执行此转换，但它对我不起作用。 datastax spark-cassandra-connector

【问题讨论】：

datastax-oss.atlassian.net/browse/SPARKC-370 - 目前 Dataframes 的行为不变。
感谢 Russ，我可以使用 RDD 解决它。

标签： scala cassandra datastax-enterprise spark-cassandra-connector

【解决方案1】：

您的 MyRow 定义似乎与 cassandra 表定义不匹配。试试这个：

val df = List((1, "My Long Description", "My Name", "My Short Name")).toDF("id", "long_name", "name", "short_name")

【讨论】：

指定列名对于像这样的简单用例来说看起来不错。但我想保存一个包含许多嵌套case classes 的复杂对象。有没有办法即时将骆驼案例属性转换为蛇案例列名称？

【解决方案2】：

使用 RDD，spark-cassandra-connector 自动将驼峰式属性转换为带下划线的列名。再次感谢RussS

这是我如何将案例类对象保存到 cassandra 表

    val writeRDD = sc.makeRDD(List(new MyRow(2,Option("long name"), "ss", Option("short name"))))
    writeRDD.saveToCassandra("dev", "castable")

【讨论】：