如何使用Java版Spark查询Cassandra表的非主键列

没问题，我来帮你搞定这个需求。结合Spark查询Cassandra的非主键列其实很直观，下面一步步给你讲清楚实现步骤和代码示例。

第一步：准备依赖 #

首先你需要在项目中引入Spark和Spark-Cassandra连接器的依赖。如果用Maven的话，把下面的依赖加到你的pom.xml里（注意根据你的Spark和Cassandra版本调整版本号）：

<dependencies>
    <!-- Spark Core -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.12</artifactId>
        <version>3.3.2</version>
    </dependency>
    <!-- Spark SQL -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.12</artifactId>
        <version>3.3.2</version>
    </dependency>
    <!-- Spark Cassandra Connector -->
    <dependency>
        <groupId>com.datastax.spark</groupId>
        <artifactId>spark-cassandra-connector_2.12</artifactId>
        <version>3.3.0</version>
    </dependency>
</dependencies>

第二步：初始化SparkSession #

要连接Cassandra，你需要在SparkSession中配置Cassandra的连接信息，比如节点地址、目标keyspace等：

import org.apache.spark.sql.SparkSession;

public class CassandraSparkQuery {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("CassandraNonPrimaryKeyQuery")
                .master("local[*]") // 本地调试用，生产环境去掉或者换成集群地址
                .config("spark.cassandra.connection.host", "你的Cassandra节点IP")
                .config("spark.cassandra.connection.port", "9042")
                // 如果Cassandra开启了认证，加上下面两行
                // .config("spark.cassandra.auth.username", "你的用户名")
                // .config("spark.cassandra.auth.password", "你的密码")
                .getOrCreate();
    }
}

第三步：加载Cassandra表并执行非主键查询 #

接下来我们用Spark SQL/DataFrame API来加载表，然后根据login和firstname过滤数据。这种方式比RDD API更简洁，也更适合数据分析场景：

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import static org.apache.spark.sql.functions.col;

// 加载Cassandra表
Dataset<Row> userDF = spark.read()
        .format("org.apache.spark.sql.cassandra")
        .option("keyspace", "mykeyspace")
        .option("table", "user")
        .load();

// 按login和firstname查询，比如找login为"harshit"且firstname为"Harshit"的用户
Dataset<Row> filteredUsers = userDF.filter(
        col("login").equalTo("harshit")
                .and(col("firstname").equalTo("Harshit"))
);

// 打印查询结果
filteredUsers.show();

// 也可以把结果转换成Java对象来处理（比如先定义一个User类）
// List<User> userList = filteredUsers.as(Encoders.bean(User.class)).collectAsList();

如果你更喜欢用SQL语法来查询，也可以先把DataFrame注册成临时视图，然后写SQL语句：

userDF.createOrReplaceTempView("user_view");

Dataset<Row> sqlResult = spark.sql(
        "SELECT * FROM user_view WHERE login = 'harshit' AND firstname = 'Harshit'"
);

sqlResult.show();

重要提示：性能优化 #

因为你查询的是非主键列，Spark默认会做全表扫描，如果你的表数据量很大，这会非常慢。这里给你两个优化建议：

1. 给非主键列建二级索引：在Cassandra中执行CREATE INDEX IF NOT EXISTS idx_user_login ON mykeyspace.user(login);和CREATE INDEX IF NOT EXISTS idx_user_firstname ON mykeyspace.user(firstname);，这样Spark查询时可以利用索引减少扫描的数据量。
2. 使用物化视图：如果经常需要按这两个列组合查询，可以创建物化视图，比如：

CREATE MATERIALIZED VIEW IF NOT EXISTS mykeyspace.user_by_login_firstname AS
SELECT id, login, password, firstname, lastname, email
FROM mykeyspace.user
WHERE login IS NOT NULL AND firstname IS NOT NULL
PRIMARY KEY((login, firstname), id);

之后Spark直接查询这个物化视图，性能会和查询主键一样快。

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