如何在将数据加载到Spark Dataset前直接在Cassandra中过滤数据（Java实现）

嗨，我完全懂你的需求——现在先把全表数据拉到Spark再过滤的方式太浪费资源了对吧？其实Spark Cassandra Connector支持过滤下推功能，能让过滤逻辑直接在Cassandra端执行，只把符合条件的数据传输给Spark，这样既能减少网络带宽消耗，也能降低Spark的内存压力，效率提升明显。

下面给你两种实现方式，都能实现“加载前过滤”的效果：

方式一：使用where()方法嵌入过滤条件 #

你可以在读取数据的链式调用中直接加入where()方法，Spark Cassandra Connector会自动把这个过滤条件转换成CQL的WHERE子句，推送给Cassandra执行：

SparkConf sparkConf = new SparkConf()
    .setMaster("local")
    .setAppName("CassandraFilterDemo")
    .set("spark.cassandra.connection.host", "localhost")
    .set("spark.cassandra.auth.username", "cassandra")
    .set("spark.cassandra.auth.password", "cassandra")
    .set("spark.cassandra.output.consistency.level", "ONE");

SparkSession spark = SparkSession.builder().config(sparkConf).getOrCreate();

// 关键变化：把过滤条件放到where()中，在load()之前执行
Dataset<Row> dataset = spark
    .read()
    .format("org.apache.spark.sql.cassandra")
    .options(ImmutableMap.of("table", "my_table", "keyspace", "my_keyspace"))
    .select("col1", "col2")
    .where("col1 > 9") // 这个条件会下推到Cassandra
    .load();

方式二：通过options直接传入CQL过滤条件 #

如果你更习惯用CQL原生的语法，也可以在options中直接指定where参数，传入完整的CQL过滤语句：

SparkConf sparkConf = new SparkConf()
    .setMaster("local")
    .setAppName("CassandraFilterDemo")
    .set("spark.cassandra.connection.host", "localhost")
    .set("spark.cassandra.auth.username", "cassandra")
    .set("spark.cassandra.auth.password", "cassandra")
    .set("spark.cassandra.output.consistency.level", "ONE");

SparkSession spark = SparkSession.builder().config(sparkConf).getOrCreate();

// 关键变化：在options中加入where参数
Dataset<Row> dataset = spark
    .read()
    .format("org.apache.spark.sql.cassandra")
    .options(ImmutableMap.of(
        "table", "my_table",
        "keyspace", "my_keyspace",
        "where", "col1 > 9" // 直接指定CQL的WHERE条件
    ))
    .select("col1", "col2")
    .load();

额外注意事项 #

• 如果col1是Cassandra表的分区键或聚类键，Cassandra能快速定位数据，过滤效率会非常高；如果是普通列，建议给该列创建二级索引，否则Cassandra会执行全表扫描，但即使是全表扫描，也比把全表数据拉到Spark再过滤要更高效。
• 确保你使用的Spark Cassandra Connector版本支持过滤下推（Spark 3.x对应的Connector 3.x及以上版本都支持这个功能）。

备注：内容来源于stack exchange，提问作者AJay