A user reports that the range query throughput is far higher than expected when setting spark.cassandra.input.readsPerSec in the spark-cassandra-connector.
Job dependencies. The Java driver version is set to 4.13.0.
<dependency>
<groupId>com.datastax.spark</groupId>
<artifactId>spark-cassandra-connector_2.12</artifactId>
<version>3.2.0</version>
<exclusions>
<exclusion>
<groupId> com.datastax.oss</groupId>
<artifactId>java-driver-core-shaded</artifactId>
</exclusion>
</exclusions>
</dependency>
...
<dependency>
<groupId>com.datastax.oss</groupId>
<artifactId>java-driver-core</artifactId>
<version>4.13.0</version>
</dependency>
There are two steps in the job (both an FTS):
Dataset<Row> dataset = sparkSession.sqlContext().read()
.format("org.apache.spark.sql.cassandra")
.option("table", "inbox_user_msg_dummy")
.option("keyspace", "ssmp_inbox2").load();
-and-
Dataset<Row> olderDataset = sparkSession.sql("SELECT * FROM inbox_user_msg_dummy where app_uuid = 'cb663e07-7bcc-4039-ae97-8fb8e8a9ff77' AND " +
"create_hour < '" + minus180DaysInstant + "'");
Job configuration:
SparkConf sparkConf = new SparkConf()
.setMaster("local[*]") //uncomment while running in local
.setAppName("inbox-gateway-spark-job")
.set("spark.scheduler.mode", "FAIR")
.set("spark.cassandra.connection.port", "9042")
.set("keyspace", "ssmp_inbox2")
.set("spark.cassandra.connection.host", "cass-556799284-1-1276056270.stg.ssmp-inbox2-stg.ms-df-cassandra.stg-az-southcentralus-6.prod.us.walmart.net,
cass-556799284-2-1276056276.stg.ssmp-inbox2-stg.ms-df-cassandra.stg-az-southcentralus-6.prod.us.walmart.net,
cass-556799284-3-1276056282.stg.ssmp-inbox2-stg.ms-df-cassandra.stg-az-southcentralus-6.prod.us.walmart.net")
.set("spark.cassandra.auth.username", "ssmp-inbox-app-v2")
.set("spark.cassandra.auth.password", "*")
.set("spark.cassandra.input.consistency.level", "LOCAL_ONE")
.set("spark.cassandra.concurrent.reads", "1")
.set("spark.cassandra.input.readsPerSec", "10")
.set("spark.cassandra.input.fetch.sizeInRows", "10")
.set("spark.cassandra.input.split.sizeInMB", "10")
.set("spark.cores.max", "20")
.set("spark.executor.memory", "20G")
.set("spark.yarn.executor.memoryOverhead", "12000")
.set("spark.cassandra.read.timeoutMS", "200000")
.set("spark.task.maxFailures", "10")
.set("spark.cassandra.connection.localDC", "southcentral");
Note that Spark is limiting the actual cores to 16 because workers have 8 cores. There is 1 executor.
When the job runs, it's observed that there are ~22k range queries/s for the first FTS nearly saturating CPU on the cluster, and for the second FTS, there are ~725 range queries/s on the table.
The expectation is that with 16 total Spark cores, the range query throughput would be limited to 160/s (spark.cassandra.input.readsPerSec * spark cores).
Is this reasoning correct? What is the recommendation for controlling read throughput from the spark-cassandra-connector?
I know we've had other users successfully configure this throttle before, but we've never looked closely at what the resulting throughput is. This one does seem to be a big discrepancy though because the two steps are essentially running the same operation - a full table scan. The queries the connector ultimately runs are the same.
The schema:
CREATE TABLE ssmp_inbox2.inbox_user_msg_dummy (
user_id text,
create_hour timestamp,
app_uuid text,
message_id text,
app_name text,
create_ts bigint,
is_actiontaken boolean,
is_compensable boolean,
is_deleted boolean,
is_read boolean,
message_payload text,
mini_app_name text,
notification text,
PRIMARY KEY ((user_id, create_hour, app_uuid), message_id)
) WITH CLUSTERING ORDER BY (message_id DESC)
AND additional_write_policy = '99p'
AND bloom_filter_fp_chance = 0.01
AND caching = {'keys': 'ALL', 'rows_per_partition': 'NONE'}
AND cdc = false
AND comment = ''
AND compaction = {'class': 'org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy', 'max_threshold': '32', 'min_threshold': '4'}
AND compression = {'chunk_length_in_kb': '64', 'class': 'org.apache.cassandra.io.compress.LZ4Compressor'}
AND crc_check_chance = 1.0
AND default_time_to_live = 0
AND extensions = {}
AND gc_grace_seconds = 864000
AND max_index_interval = 2048
AND memtable_flush_period_in_ms = 0
AND min_index_interval = 128
AND read_repair = 'BLOCKING'
AND speculative_retry = '99p';
The query:
SELECT * FROM ssmp_inbox2.inbox_user_msg_dummy WHERE token(user_id, create_hour, app_uuid) >= token(G9e7Y4Y, 2023-08-10T04:17:27.234Z, cb663e07-7bcc-4039-ae97-8fb8e8a9ff77) AND token(user_id, create_hour, app_uuid) <= 9121832956220923771 LIMIT 10
FWIW, avg partition size is 649 bytes, max is 2.7kb.
