用于Cassandra的Spark2会话,sql查询

Question

在Spark-2.0中,创建Spark会话的最佳方法是什么.因为在Spark-2.0和Cassandra中,API都经过了重新设计,基本上不赞成使用SqlContext(以及CassandraSqlContext).因此,为了执行SQL,我要么创建一个Cassandra会话(com.datastax.driver.core.Session) and use execute( " ").或者我必须创建一个SparkSession (org.apache.spark.sql.SparkSession) and execute sql(String sqlText)方法.

我不知道两者的SQL限制 - 有人可以解释一下.

此外,如果我必须创建SparkSession - 我该怎么做 - 找不到任何合适的例子.随着API的重新设计,旧示例不起作用.我正在通过这个代码示例 - DataFrames - 不清楚这里使用的是什么sql上下文(是正确的方法继续.)(由于某些原因,弃用的API甚至没有编译 - 需要检查我的eclipse设置)

谢谢

Answer 1

您需要Cassandra Session来创建/删除Cassandra DB中的密钥空间和表.在Spark应用程序中,为了创建Cassandra Session,您需要将SparkConf传递给CassandraConnector.在Spark 2.0中,您可以像下面这样做.

 SparkSession spark = SparkSession
              .builder()
              .appName("SparkCassandraApp")
              .config("spark.cassandra.connection.host", "localhost")
              .config("spark.cassandra.connection.port", "9042")
              .master("local[2]")
              .getOrCreate();

CassandraConnector connector = CassandraConnector.apply(spark.sparkContext().conf());
Session session = connector.openSession();
session.execute("CREATE TABLE mykeyspace.mytable(id UUID PRIMARY KEY, username TEXT, email TEXT)");

如果您有现有的Dataframe,那么您也可以使用Cassandra创建表DataFrameFunctions.createCassandraTable(Df).见API的细节在这里.

您可以使用spark-cassandra-connector提供的api从Cassandra DB读取数据,如下所示.

Dataset<Row> dataset = spark.read().format("org.apache.spark.sql.cassandra")
            .options(new HashMap<String, String>() {
                {
                    put("keyspace", "mykeyspace");
                    put("table", "mytable");
                }
            }).load();

dataset.show(); 

您可以使用SparkSession.sql()方法在spark cassandra连接器返回的Dataframe上创建的临时表上运行查询,如下所示.

dataset.createOrReplaceTempView("usertable");
Dataset<Row> dataset1 = spark.sql("select * from usertable where username = 'Mat'");
dataset1.show();