如何从Spark Cassandra Connector读取指标并排查读写Cassandra耗时（2.0.1版本）

Hey there! Let's break down how you can get metrics (including table write latencies) and measure Spark's read/write times for Cassandra when using spark-cassandra-connector_2.11-2.0.1. I’ve worked with this specific version a few times, so here’s a practical breakdown:

The 2.0.1 connector uses Dropwizard Metrics under the hood, which integrates seamlessly with Spark’s MetricsSystem. Here’s how to tap into it:

• First, confirm metrics are enabled
  By default, connector metrics should be turned on, but you can explicitly set this in your spark-defaults.conf or via spark-submit flags to be safe:

  spark.cassandra.metrics.enabled=true
  

• Key metrics for write operations
  You’ll find granular metrics for both global and table-specific writes:

  • cassandra.driver.metrics.session.Writes.total: Total number of write requests sent to Cassandra
  • cassandra.driver.metrics.session.Writes.latency.mean: Average latency per write request (in milliseconds)
  • cassandra.driver.metrics.session.Writes.latency.max: Maximum latency for any write request
  • Table-specific metrics: cassandra.driver.metrics.session.Writes.[your_keyspace].[your_table].latency.* (replace placeholders with your actual keyspace/table names)

• Retrieve metrics in code
  You can pull these metrics directly from Spark’s MetricsSystem in your application code. Here’s a Scala example:

  import org.apache.spark.SparkContext
  import org.apache.spark.metrics.source.Source
  
  // Assuming you have an existing SparkContext
  val sc: SparkContext = spark.sparkContext
  // Get all metrics sources tied to Cassandra
  val cassandraMetrics = sc.env.metricsSystem.getSourcesByName("cassandra")
  
  cassandraMetrics.foreach { source =>
    println("=== Cassandra Connector Metrics ===")
    source.metrics.foreach { case (metricName, metric) =>
      println(s"$metricName: ${metric.getValue}")
    }
  }
  

  Alternatively, if you’ve configured a metrics sink (like console or JMX), these metrics will show up in Spark’s UI under the "Metrics" tab too.

Beyond connector-specific metrics, you’ll want to measure the end-to-end time Spark spends reading from or writing to Cassandra. Here are two easy ways:

• Use the Spark UI
  This is the quickest no-code approach:

  1. After submitting your job, open the Spark UI (usually at http://<driver-ip>:4040)
  2. For write operations: Navigate to "Jobs" → find the stage corresponding to your Cassandra write (look for CassandraSourceRelation in the stage description) → check the "Duration" field for total time, or drill into individual tasks to see per-task latency.
  3. For read operations: If using Spark SQL, check the "SQL" tab to see the total query duration. For RDD/DataFrame reads, find the stage that loads data from Cassandra and check its duration.

• Manual timing in your code
  If you need to log or programmatically use the latency values, add simple timing around your read/write actions. Remember: Spark is lazy, so reads won’t execute until you trigger an action (like count() or collect()). Example:

  // Measure write time
  val writeStartTime = System.currentTimeMillis()
  yourDataFrame.write
    .format("org.apache.spark.sql.cassandra")
    .options(Map("table" -> "your_table", "keyspace" -> "your_keyspace"))
    .mode(org.apache.spark.sql.SaveMode.Append)
    .save()
  val writeEndTime = System.currentTimeMillis()
  println(s"Total write time: ${(writeEndTime - writeStartTime)/1000.0} seconds")
  
  // Measure read time (with action to trigger execution)
  val readStartTime = System.currentTimeMillis()
  val cassandraDF = spark.read
    .format("org.apache.spark.sql.cassandra")
    .options(Map("table" -> "your_table", "keyspace" -> "your_keyspace"))
    .load()
  // Trigger an action to actually read the data
  val recordCount = cassandraDF.count()
  val readEndTime = System.currentTimeMillis()
  println(s"Total read time (including count action): ${(readEndTime - readStartTime)/1000.0} seconds")
  

If you want to monitor metrics live (without digging into logs or code), enable JMX for your Spark driver and executors:
Add these flags when submitting your job:

spark-submit \
  --conf spark.driver.extraJavaOptions="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=9999 -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false" \
  --conf spark.executor.extraJavaOptions="-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=9998 -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false" \
  your-application.jar

Then use tools like JConsole or VisualVM to connect to the JMX ports. Look for MBeans under com.datastax.driver.core—you’ll find all the connector’s metrics, including per-table write latencies, request rates, and error counts.

内容的提问来源于stack exchange，提问作者Soumya Ranjan Mahunta