在以前的Spark版本中,如1.6.1,我正在使用Spark Context创建Cassandra Context,
import org.apache.spark.{ Logging, SparkContext, SparkConf }
//config
val conf: org.apache.spark.SparkConf = new SparkConf(true)
.set("spark.cassandra.connection.host", CassandraHost)
.setAppName(getClass.getSimpleName)
lazy val sc = new SparkContext(conf)
val cassandraSqlCtx: org.apache.spark.sql.cassandra.CassandraSQLContext = new CassandraSQLContext(sc)
//Query using Cassandra context
cassandraSqlCtx.sql("select id from table ")
但是在Spark 2.0中,Spark Context被Spark session取代了,我怎么使用cassandra Context ?
简短的回答:你没有。它已被弃用并删除。
长答案:你不想。HiveContext提供了除目录之外的所有内容,并支持范围更广的SQL(HQL~)。在Spark 2.0中,这意味着你需要使用createOrReplaceTempView手动注册Cassandra表,直到实现externalcatalog。
在Sql中看起来像
spark.sql("""CREATE TEMPORARY TABLE words
|USING org.apache.spark.sql.cassandra
|OPTIONS (
| table "words",
| keyspace "test")""".stripMargin)
在原始DF api中,它看起来像
spark
.read
.format("org.apache.spark.sql.cassandra")
.options(Map("keyspace" -> "test", "table" -> "words"))
.load
.createOrReplaceTempView("words")
这两个命令都将为SQL查询注册表"words"