如何使用 PySpark、SparkSQL 和 Cassandra？

Posted 2023-04-17

【中文标题】如何使用 PySpark、SparkSQL 和 Cassandra？

【英文标题】：How to work with PySpark, SparkSQL and Cassandra?

【发布时间】：2016-02-28 16:06:26

【问题描述】：

我对这个故事中的不同参与者有点困惑：PySpark、SparkSQL、Cassandra 和 pyspark-cassandra 连接器。

据我了解，Spark 发展了很多，SparkSQL 现在是一个关键组件（带有“数据帧”）。显然，没有 SparkSQL 绝对没有理由工作，尤其是在连接到 Cassandra 时。

所以我的问题是：需要什么组件以及如何以最简单的方式将它们连接在一起？

在 Scala 中使用 spark-shell 我可以简单地完成

./bin/spark-shell --jars spark-cassandra-connector-java-assembly-1.6.0-M1-SNAPSHOT.jar

然后

import org.apache.spark.sql.cassandra.CassandraSQLContext
val cc = new CassandraSQLContext(sc)
cc.setKeyspace("mykeyspace")
val dataframe = cc.sql("SELECT count(*) FROM mytable group by beamstamp")

我如何使用pyspark 做到这一点？

这里有几个子问题以及我收集的部分答案（如果我错了，请纠正）。

是否需要 pyspark-casmandra（我不这么认为 - 我一开始不明白在做什么）

我需要使用pyspark 还是我可以使用我的常规jupyter notebook 并自己导入必要的东西？

【参考方案1】：

Pyspark 应该使用 spark-cassandra-connector 包启动，如 Spark Cassandra 连接器 python docs 中所述。

./bin/pyspark 
  --packages com.datastax.spark:spark-cassandra-connector_$SPARK_SCALA_VERSION:$SPARK_VERSION

加载后，您将能够在 C* 数据帧上使用 Spark 中已经存在的任何数据帧操作。 More details on options of using C* dataframes.

要将其设置为使用jupyter notebook 运行，只需使用以下属性设置您的环境。

export PYSPARK_DRIVER_PYTHON=ipython
export PYSPARK_DRIVER_PYTHON_OPTS=notebook

调用pyspark 将启动正确配置的笔记本。

没有必要使用pyspark-cassandra，除非你对在 python 中使用 RDD 感兴趣，因为它有一些性能缺陷。

【参考方案2】：

在 Python 连接器中公开了 DataFrame API。只要spark-cassandra-connector 可用并且SparkConf 包含所需的配置，就不需要额外的包。您可以简单地指定格式和选项：

df = (sqlContext
    .read
    .format("org.apache.spark.sql.cassandra")
    .options(table="mytable", keyspace="mykeyspace")
    .load())

如果您想使用纯 SQL，您可以注册DataFrame，如下所示：

df.registerTempTable("mytable")

## Optionally cache
sqlContext.cacheTable("mytable")

sqlContext.sql("SELECT count(*) FROM mytable group by beamstamp")

连接器的高级功能，例如 CassandraRDD 不会暴露给 Python，因此如果您需要超出 DataFrame 功能的功能，那么 pyspark-cassandra 可能会很有用。

【讨论】：

好的，谢谢。是否有可能（如在 Scala 中）通过 SQL api 直接访问表？

据我所知这是不可能的。你当然可以像往常一样registerTempTable。此外，没有 CassandraRDD 的等价物。

好的。稍后我会发布一些代码以确保我做对了。