如何在PySpark中执行CQL？已实现Cassandra表读写，寻求操作方案

嘿，你已经搞定了PySpark和Cassandra的基础读写，接下来要执行CQL语句的话，有几个靠谱的方案，我给你一一拆解：

方法一：用CassandraConnector执行原生CQL #

你当前使用的anguenot/pyspark-cassandra包已经封装了Cassandra的底层连接能力，可以直接通过CassandraConnector来执行任意CQL语句，步骤很清晰：

1. 先导入需要的类：

from pyspark_cassandra import CassandraConnector

2. 创建连接器实例——它会自动读取你启动PySpark时配置的spark.cassandra.connection.host等参数：

connector = CassandraConnector(sc)

3. 通过withSessionDo获取Cassandra Session，然后执行CQL：

# 执行建表的DDL语句
connector.withSessionDo(lambda session: session.execute("CREATE TABLE IF NOT EXISTS my_keyspace.new_table (id INT PRIMARY KEY, name TEXT);"))

# 执行插入数据的DML语句
connector.withSessionDo(lambda session: session.execute("INSERT INTO my_keyspace.new_table (id, name) VALUES (1, 'test_data');"))

# 执行查询并遍历结果
query_result = connector.withSessionDo(lambda session: session.execute("SELECT * FROM my_keyspace.new_table;"))
for row in query_result:
    print(f"ID: {row.id}, Name: {row.name}")

这种方法特别适合执行DDL（建表、删表）或者复杂的原生CQL操作，能直接和Cassandra的Session交互，灵活性拉满。

方法二：通过Spark SQL兼容CQL查询（适合数据查询场景） #

如果你的需求是查询数据并转换成Spark DataFrame，可以先把Cassandra表注册成临时视图，再用Spark SQL执行类CQL的查询（语法和CQL基本兼容，少量差异）：

1. 注册Cassandra表为临时视图：

sqlContext.read\
 .format("org.apache.spark.sql.cassandra")\
 .options(table="my_table", keyspace="my_keyspace")\
 .load()\
 .createOrReplaceTempView("my_cassandra_view")

2. 用Spark SQL执行查询：

result_df = sqlContext.sql("SELECT cCode, mDate FROM my_cassandra_view WHERE pCode = 31")
result_df.show()

不过这种方式更偏向数据查询，要是你需要执行修改类的CQL（比如插入、删除），还是第一种方法更直接高效。

小提醒 #

• 确保pyspark-cassandra版本和你的Spark版本兼容，你用的0.7.0版本适配Spark 2.x，这点没问题；
• 执行CQL时要确认Spark节点能访问Cassandra集群，且有对应的操作权限；
• 批量写入数据优先用Spark DataFrame API，比逐条执行CQL效率高很多，特殊原生CQL需求再用第一种方法。

内容的提问来源于stack exchange，提问作者Steven