In Cassandra, how to select rows based on a column and also update that column later

user_405855 0
Viewed 93

I am new to cassandra database design. My app is an orchestrator service that stores requests come from other apps (into 'inbound' table). After that, there will be a cron job that picks up rows in table and process them.

-inbound
    request_id     uuid
    payload        varchar    
    status_code    text        will be one of those values {RECEIVED, IN_PROCESS, FINISHED, FAILED}
    tryNumber      int         start with 0 and be increased for every try
    update_date    timestamp
    create_date    timestamp

Here are the queries that will be applied to this table:

1. insert request into this table, status=RECEIVED, create_date = update_date = current timestamp
2. pickup the first 100 (or ALL) requests with status = RECEIVED or FAILED to process.
3. for each pick-up request, update status to PROCESS and increase tryNumber and  update_date = current time
4. after processing, for each pick-up request, update status to FINISHED or FAILED, and also update_date = current time

I have tried to make status_code (partition key) and request_id(cluster key) to make it able to do 2nd requirement, but then it is impossible to do 3rd and 4th requirements.

I also hear that denormalize can help but i dont know how to to that.

Could you please help me on this design using cassandra database.

denormalization nosql composite-primary-key cassandra database-design
1 Answers

So you'll want to build this model as a time series with a time component "bucket" to keep the partitions from growing too large. To do that, we'll need one more column: month_bucket.

The "bucket" column is dependent on two things:

  1. Your business logic. Do you typically query for data within a certain month? Or week? Or does the job process it fast enough that you really only need to worry about the current day?
  2. Number of rows written over time. Cassandra has a cells/per-partition limit of 2 billion, size of 2GB, and the partition will become unusable long before those hard limits. I recommend keeping no more than 20k-50k rows in a partition, and less than 10MB.

The point, is to take this example with "a grain of salt." A different time bucket may make more sense for you (week, day, etc.). I've built models using a bucket based on hour before, because of the expected write throughput.

So month_bucket will be the partition key, and we'll cluster by create_date in ASCending order. That way, all the most-recently created rows will be at the "bottom" of the partition, and you'll have an easier time with queries like "get the oldest 100 w/ a certain status" (which will read from the "top" of the partition).

To query by status_code, you'll need a secondary index on it. Yes, I'm sure you've read "secondary indexes are bad." And that's true. But, in certain cases with (what I call) "middle of the road" cardinality, they are ok. They are particularly useful with a finite amount of options (processed, received, etc.) on a column which can change (can't change the values of primary key components). But with the queries you describe, you'll be using a secondary index with the partition keys, which will limit their "badness" significantly.

In short, I'd build the table like this:

CREATE TABLE inbound (
  request_id     uuid,
  payload        text,
  status_code    text,
  tryNumber      int,
  update_date    timestamp,
  create_date    timestamp,
  month_bucket   int,
  PRIMARY KEY (month_bucket,create_date,request_id)
) WITH CLUSTERING ORDER BY (create_date ASC, request_id ASC);

CREATE INDEX ON inbound (status_code);

With this PRIMARY KEY definition, all rows will be stored by month (ex: 202110), ordered by create_date, and with request_id to help ensure uniqueness.

My only concerns here, are around the use of tryNumber:

  1. Cassandra doesn't overwrite. It stores obsoleted values for up to 10 days. Too many updates, and there will be a loooooong list of obsoleted values that Cassandra has to sort through, not to mention read from multiple files on-disk. But if the idea here is that it typically only gets updated once or twice, it'll be ok.
  2. Cassandra doesn't provide ACID transactions, which makes multi-threaded operations all doing read-write-read-write a bit of an anti-pattern. There's no way to "lock" a value to prevent it from being updated by another process (race condition). Ex: the job reads a tryNumber of 0 and attempts to set it to 1, but another process sets it to 1 before that happens. Just something to be aware of.

With this model, queries like this should work just fine:

SELECT * FROM inbound
WHERE month_bucket=202110 AND status_code='RECEIVED'
LIMIT 100;

Or:

SELECT * FROM inbound
WHERE month_bucket=202110 AND status_code IN ('RECEIVED','FAILED')
LIMIT 100;

Again, with the rows in ascending order by create_date, this should give you the oldest 100 with a particular status. Cassandra does't have an OR keyword, because that incurs "random" reads, and Cassandra was built for "sequential" reads. Using IN is a way around that, but use of IN comes with the same precautions that secondary indexes have (it's not too bad if you're using it while filtering on a partition key).

user_10545580
Related