scala – 使用Spark DataFrame将数据插入Cassandra表

发布时间：2020-12-16 09:58:51 所属栏目：安全来源：网络整理

导读：我正在使用 Scala版本2.10.5 Cassandra 3.0和Spark 1.6.我想将数据插入cassandra所以我尝试了基本的例子 scala val collection = sc.parallelize(Seq(("cat",30),("fox",40)))scala collection.saveToCassandra("test","words",SomeColumns("word","count"))

我正在使用 Scala版本2.10.5 Cassandra 3.0和Spark 1.6.我想将数据插入cassandra所以我尝试了基本的例子

scala> val collection = sc.parallelize(Seq(("cat",30),("fox",40)))
scala> collection.saveToCassandra("test","words",SomeColumns("word","count"))

哪些Works和能够将数据插入到Cassandra中.所以我有一个csv文件,我想通过匹配模式插入到Cassandra表中

val person = sc.textFile("hdfs://localhost:9000/user/hduser/person")
import org.apache.spark.sql._
val schema =  StructType(Array(StructField("firstName",StringType,true),StructField("lastName",StructField("age",IntegerType,true)))
val rowRDD = person.map(_.split(",")).map(p =&gt; org.apache.spark.sql.Row(p(0),p(1),p(2).toInt))
val personSchemaRDD = sqlContext.applySchema(rowRDD,schema)
 personSchemaRDD.saveToCassandra

当我使用SaveToCassndra时,我得到saveToCassandra不是personSchemaRDD的一部分.所以教会以不同的方式尝试

df.write.format("org.apache.spark.sql.cassandra").options(Map( "table" -> "words_copy","keyspace" -> "test")).save()

但是无法连接到ip上的cassandra：port.can任何人告诉我最好的方法.我需要定期从文件中将数据保存到cassandra.

解决方法

sqlContext.applySchema(…)返回一个DataFrame,而DataFrame没有saveToCassandra方法.

你可以使用它的.write方法：

val personDF = sqlContext.applySchema(rowRDD,schema)
personDF.write.format("org.apache.spark.sql.cassandra").options(Map( "table" -> "words_copy","keyspace" -> "test")).save()

如果我们想使用savetoCassandra方法,最好的方法是使用case类来获得模式感知的RDD.

case class Person(firstname:String,lastName:String,age:Int)
val rowRDD = person.map(_.split(",")).map(p => Person(p(0),p(2).toInt)
rowRDD.saveToCassandra(keyspace,table)

Dataframe写方法应该有效.检查您是否正确配置了上下文.

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!