Scala/Spark/Cassandra项目：声明SparkContext及使用连接器访问Cassandra表

嘿，作为Scala和Spark的新手，你已经开了个好头啦！我帮你把这个Cassandra存储类补全并优化，同时给你讲讲每个部分的作用，方便你上手：

完整实现代码 #

import org.apache.spark.sql.cassandra._
import com.datastax.spark.connector._
import org.apache.spark.SparkContext
import com.datastax.spark.connector.ColumnSelector
import org.apache.spark.sql.SparkSession

// 定义通用存储接口，方便后续扩展其他存储类型
trait Storage {
  def readTable(tableName: String): Unit
  def writeToTable[T <: Product](tableName: String, data: Seq[T]): Unit
}

// Cassandra具体实现类
class CassandraStorage(sc: SparkContext, keyspaceName: String) extends Storage {

  // 读取Cassandra表并打印前10条数据（新手快速验证用）
  override def readTable(tableName: String): Unit = {
    val cassandraRDD = sc.cassandraTable(keyspaceName, tableName)
    println(s"=== 读取${keyspaceName}.${tableName}的前10条数据 ===")
    cassandraRDD.take(10).foreach(println)
  }

  // 将Case Class集合写入Cassandra（最常用的写入方式）
  override def writeToTable[T <: Product](tableName: String, data: Seq[T]): Unit = {
    sc.parallelize(data).saveToCassandra(keyspaceName, tableName)
    println(s"=== 成功写入${data.size}条数据到${keyspaceName}.${tableName} ===")
  }

  // 扩展：用Spark SQL读取表（适合数据分析场景）
  def readTableWithSparkSQL(tableName: String): Unit = {
    val spark = SparkSession.builder().getOrCreate()
    val df = spark.read
      .format("org.apache.spark.sql.cassandra")
      .options(Map("keyspace" -> keyspaceName, "table" -> tableName))
      .load()

    println(s"=== ${keyspaceName}.${tableName}的表结构 ===")
    df.printSchema()
    println(s"=== ${keyspaceName}.${tableName}的数据预览 ===")
    df.show(10)
  }
}

// 测试示例（直接运行就能验证功能）
object CassandraStorageDemo {
  def main(args: Array[String]): Unit = {
    // 初始化SparkContext（本地测试用local[*]，生产环境要改对应master）
    val sc = new SparkContext("local[*]", "CassandraStorageTest")
    // 配置Cassandra连接地址（替换成你的集群地址）
    sc.set("spark.cassandra.connection.host", "127.0.0.1")

    // 创建存储实例
    val storage = new CassandraStorage(sc, "your_keyspace")

    // 1. 测试读取表
    storage.readTable("your_table")

    // 2. 测试写入数据（先定义和表结构匹配的Case Class）
    case class User(user_id: Int, user_name: String, email: String)
    val testUsers = Seq(
      User(1, "Alice", "alice@test.com"),
      User(2, "Bob", "bob@test.com")
    )
    storage.writeToTable("user", testUsers)

    // 3. 测试Spark SQL读取
    storage.readTableWithSparkSQL("user")

    // 关闭资源
    sc.stop()
  }
}

关键知识点解释 #

• Storage Trait的作用：它是一个通用接口，定义了读写的标准方法。以后如果要切换到HBase或者其他存储，只需要重新实现这个Trait就行，不用改上层业务代码，这是面向接口编程的小技巧~
• SparkContext的重要性：它是Spark应用的核心入口，spark-cassandra-connector靠它来和Cassandra集群建立连接，所以构造方法里必须传入它。
• 核心API用法：
  • sc.cassandraTable：返回对应表的RDD，适合做底层的分布式操作；
  • saveToCassandra：把RDD数据写入Cassandra，要求数据是Case Class或者Tuple，字段名必须和Cassandra表的列名完全一致（Cassandra默认列名是小写，注意别写错）；
  • Spark SQL方式：用format("org.apache.spark.sql.cassandra")加载表，得到DataFrame后可以用SQL语句查询，数据分析场景下更方便。
• 新手必踩坑提示：
  • 一定要保证spark-cassandra-connector的版本和你的Spark、Cassandra版本匹配！比如Spark 3.3对应connector 3.3.x，版本不兼容会报各种奇怪的错；
  • 必须配置spark.cassandra.connection.host，如果Cassandra有多个节点，可以用逗号分隔多个地址；
  • 本地测试的时候用local[*]作为Spark的master，这样会自动使用所有可用的CPU核心，避免资源不足。

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