Cassandra data model for application logs (billions of operations!)

Question

Say, I want to collect logs from a huge application cluster which produces 1000-5000 records per second. In future this number might reach 100000 records per second, aggregated from a 10000-strong datacenter.

CREATE TABLE operation_log (
       -- Seconds will be used as row keys, thus each row will
       -- contain 1000-5000 log messages.
       time_s bigint,
       time_ms int, -- Microseconds (to sort data within one row).
       uuid uuid, -- Monotonous UUID (NOT time-based UUID1)

       host text,
       username text,
       accountno bigint,
       remoteaddr inet,
       op_type text,
       -- For future filters — renaming a column must be faster
       -- than adding a column?
       reserved1 text,
       reserved2 text,
       reserved3 text,
       reserved4 text,
       reserved5 text,

       -- 16*n bytes of UUIDs of connected messages, usually 0,
       -- sometimes up to 100.
       submessages blob,
       request text,
       PRIMARY KEY ((time_s), time_ms, uuid)) -- Partition on time_s
-- Because queries will be "from current time into the past"
WITH CLUSTERING ORDER BY (time_ms DESC)

CREATE INDEX oplog_remoteaddr ON operation_log (remoteaddr);
...
(secondary indices on host, username, accountno, op_type);
...
CREATE TABLE uuid_lookup (
       uuid uuid,
       time_s bigint,
       time_ms int,
       PRIMARY KEY (uuid));

I want to use OrderedPartitioner which will spread data all over the cluster by its time_s (seconds). It must also scale to dozens of concurrent data writers as more application log aggregators are added to the application cluster (uniqueness and consistency is guaranteed by the uuid part of the PK).

Analysts will have to look at this data by performing these sorts of queries:

• range query over time_s, filtering on any of the data fields (SELECT * FROM operation_log WHERE time_s < $time1 AND time_s > $time2 AND $filters),
• pagination query from the results of the previous one (SELECT * FROM operation_log WHERE time_s < $time1 AND time_s > $time2 AND token(uuid) < token($uuid) AND $filters),
• count messages filtered by any data fields within a time range (SELECT COUNT(*) FROM operation_log WHERE time_s < $time1 AND time_s > $time2 AND $filters),
• group all data by any of the data fields within some range (will be performed by application code),
• request dozens or hundreds of log messages by their uuid (hundreds of SELECT * FROM uuid_lookup WHERE uuid IN [00000005-3ecd-0c92-fae3-1f48, ...]).

My questions are:

• Is this a sane data model?
• Is using OrderedPartitioner the way to go here?
• Does provisioning a few columns for potential filter make sense? Or is adding a column every once in a while cheap enough to run on a Cassandra cluster with some reserved headroom?
• Is there anything that prevents it from scaling to 100000 inserted rows per second from hundreds of aggregators and storing a petabyte or two of queryable data, provided that the number of concurrent queryists will never exceed 10?

Answer 1

This data model is close to a sane model, with several important modifications/caveats:

• Do not use ByteOrderedPartitioner, especially not with time as the key. Doing this will result in severe hotspots on your cluster, as you'll do most of your reads and all your writes to only part of the data range (and therefore a small subset of your cluster). Use Murmur3Partitioner.

• To enable your range queries, you'll need a sentinel key--a key you can know in advance. For log data, this is probably a time bucket + some other known value that's not time-based (so your writes are evenly distributed).

• Your indices might be ok, but it's hard to tell without knowing your data. Make sure your values are low in cardinality, or the index won't scale well.

• Make sure any potential filter columns adhere to the low cardinality rule. Better yet, if you don't need real-time queries, use Spark to do your analysis. You should create new columns as needed, as this is not a big deal. Cassandra stores them sparsely. Better yet, if you use Spark, you can store these values in a map.

• If you follow these guidelines, you can scale as big as you want. If not, you will have very poor performance and will likely get performance equivalent to a single node.