如何使用PySpark对Cassandra嵌套行值执行SQL聚合优化查询？

我来给你梳理一套可行的方案，刚好我之前处理过类似的嵌套JSON+Cassandra+PySpark的场景，完全能解决你现在的痛点——不用再手动发数千次查询，还能轻松用标准SQL做聚合。

一、先搞定PySpark与Cassandra的连接 #

首先得确保你的环境里装了spark-cassandra-connector，这是连接两者的核心依赖。启动PySpark的时候直接带上依赖包就行：

pyspark --packages com.datastax.spark:spark-cassandra-connector_2.12:3.4.1

如果是用Jupyter或者脚本运行，也可以在代码里配置SparkSession的时候指定依赖。接下来读取Cassandra表，Spark会自动帮你做并行处理，不用自己写循环发查询：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("CassandraNestedJSONProcessing") \
    .config("spark.cassandra.connection.host", "你的Cassandra主机地址") \
    .config("spark.cassandra.connection.port", "9042") \
    .config("spark.cassandra.auth.username", "你的用户名") \
    .config("spark.cassandra.auth.password", "你的密码") \
    .getOrCreate()

# 替换成你的keyspace和表名
df = spark.read \
    .format("org.apache.spark.sql.cassandra") \
    .options(table="目标表名", keyspace="目标keyspace") \
    .load()

二、自定义映射：把嵌套JSON转成扁平列结构 #

这一步是核心，要把嵌套的attributes字段拆成Spark能识别的列。分两种情况处理：

情况1：attributes是JSON字符串类型 #

如果Cassandra里存的是纯JSON字符串，先用from_json配合自定义Schema解析成Struct类型：

from pyspark.sql.functions import from_json
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# 完全按照你的JSON结构定义Schema，比如包含company_id、page_visits和深层嵌套
attributes_schema = StructType([
    StructField("company_id", StringType(), nullable=True),
    StructField("page_visits", IntegerType(), nullable=True),
    StructField("user_info", StructType([
        StructField("contact", StructType([
            StructField("city", StringType(), nullable=True)
        ]))
    ]))
])

# 解析JSON字符串为结构化数据
df_parsed = df.withColumn("attributes_struct", from_json(df.attributes, attributes_schema))

情况2：attributes已经是Cassandra的UDT/嵌套结构 #

这种更简单，直接用.操作符提取字段就行，还可以用selectExpr批量展开：

# 提取需要的字段，包括深层嵌套的子字段
df_flattened = df.select(
    df.attributes.company_id.alias("company_id"),
    df.attributes.page_visits.alias("page_visits"),
    df.attributes.user_info.contact.city.alias("user_city"),
    # 保留其他原始字段
    "其他需要的字段"
)

如果嵌套特别深，手动写太麻烦，还可以写个递归函数自动扁平化所有嵌套字段：

from pyspark.sql.types import StructType

def flatten_struct(df, struct_col_name, prefix=""):
    fields = []
    struct_schema = df.schema[struct_col_name].dataType
    for field in struct_schema.fields:
        full_col_name = f"{prefix}{field.name}"
        if isinstance(field.dataType, StructType):
            # 递归处理嵌套结构
            sub_fields = flatten_struct(df.select(f"{struct_col_name}.{field.name}"), field.name, prefix=f"{full_col_name}_")
            fields.extend(sub_fields)
        else:
            fields.append(f"{struct_col_name}.{field.name} AS {full_col_name}")
    return df.selectExpr(*fields)

# 调用函数自动扁平化attributes字段
df_flattened = flatten_struct(df, "attributes")

三、注册临时视图，执行SQL聚合 #

现在数据已经是扁平的列结构了，注册成临时视图就能用标准SQL写聚合逻辑，完全不用碰CQL：

# 注册临时视图，方便用SQL查询
df_flattened.createOrReplaceTempView("flattened_cassandra_data")

# 举个例子：按company_id统计总访问量和用户所在城市数
agg_result = spark.sql("""
    SELECT 
        company_id,
        SUM(page_visits) AS total_page_visits,
        COUNT(DISTINCT user_city) AS unique_cities
    FROM flattened_cassandra_data
    WHERE company_id IS NOT NULL
    GROUP BY company_id
    ORDER BY total_page_visits DESC
""")

# 查看结果
agg_result.show()

# 如果需要把结果写回Cassandra，直接用Spark的数据源即可
agg_result.write \
    .format("org.apache.spark.sql.cassandra") \
    .options(table="聚合结果表", keyspace="目标keyspace") \
    .mode("overwrite") \
    .save()

四、几个实用优化技巧 #

• 调整并行度：如果Cassandra集群节点多，可以设置spark.cassandra.input.split.size_in_mb（比如设为64），让Spark生成更多分区，充分利用集群资源：

  spark.conf.set("spark.cassandra.input.split.size_in_mb", "64")
  

• 提前过滤数据：如果不需要全表数据，读取的时候就加过滤条件，减少后续处理的数据量：

  df = spark.read \
      .format("org.apache.spark.sql.cassandra") \
      .options(table="目标表名", keyspace="目标keyspace") \
      .load() \
      .filter("page_visits > 0")
  

• 缓存中间结果：如果要多次操作扁平化后的DataFrame，记得缓存起来，避免重复解析嵌套结构：

  df_flattened.cache()
  

这样一套流程下来，不仅能替代之前的数千次单条查询，还能借助Spark的并行能力大幅提升处理效率，用SQL做聚合也比手动写逻辑灵活太多！

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