Cassandra 用于存储点击日志

Question

我从事广告技术工作，我们当前的基础架构使用 MySQL 来存储点击和转化日志。到目前为止，MySQL 对我们运行针对点击数据的即席查询很有用。 我们正在考虑切换到 Cassandra，因为我们在高峰时段会收到巨大的流量高峰。不仅如此，我们还在以非常快的速度增长，我们时不时地每秒获得大约 500-1000 次点击（持续时间较长，有时持续 20-30 分钟）。 我一直是可用的选项，到目前为止，我的研究让我相信在写入性能方面没有什么能比 Cassandra 更好。 我目前正在创建一个数据模型来存储点击。 任何点击的主要组成部分如下：

1. 广告系列 ID
2. 发布号
3. 时间戳
4. 广告素材 ID
5. 事件代码（是有效点击还是无效点击。这是一个 int 值。例如，event_code=0 是有效点击）

现在，我需要支持以下查询：

 1. SELECT * FROM clicks WHERE campaign_id=?
 2. SELECT * FROM clicks WHERE campaign_id=? AND date_time>=? AND date_time <=? 
 3. SELECT * FROM clicks WHERE campaign_id=? AND pub_id=? AND  AND date_time>=? AND date_time <=?  AND event_code=?

等 这对 MySQL 来说很简单，之后我只需从 CSV 文件中的这些查询中获取所有数据。 但是，如果我要根据第一个查询对表进行建模，这意味着我需要在 Cassandra 中创建一个表，如下所示：

    CREATE TABLE clicks_by_campaign(
     camp_id int,
     pub_id int,
     date_time timestamp,
     creative_id int,
     event_code int,
    //other fields like ip, user agent ,device etc,
    PRIMARY KEY(camp_id,pub_id,date_time,event_code,creative_id))

但有些广告系列可能有数百万行。例如，我们有具有特定 id 的广告系列，比如 id=3，点击次数超过 700 万。 这不会造成宽行问题吗？据我了解，所有这些活动数据都将作为一个分区存储在一台物理机器上。我的想法是正确的还是我错过了什么？请注意，还必须支持其他查询。例如，我可能必须共享特定发布者的点击日志（与广告活动 ID 无关）。在这种情况下，查询将如下所示：

SELECT * FROM clicks_by_publisher WHERE pub_id=? 

这显然意味着我必须创建另一个名为“clicks_by_publisher”等的表。

我还想指出，我将使用 Apache Flink 在 1 分钟的时间窗口内分析、汇总和分组点击信息。这些结果将进一步存储到 MySQL 中，以尽可能多地支持 ad-hoc 查询。

谁能指出我正确的方向。 还有其他我可以使用的策略吗？我错过了什么吗？

Answer 1

您有几个选择。我觉得我可以描述的三个。首先是指定列如下

campaign_id = PRIMARY_KEY
event_code = CLUSTER_KEY
date_time = CLUSTER_KEY

可以对集群键运行大于或等于查询。您的查询将运行。

您说得对，这将为每个广告系列 ID 创建一个分区。要解决将行存储在一台物理机器上的问题，您可以创建一个不同的表，将活动 ID 链接到 clicks 表中的行 ID。这将减少存储在单台机器上的整体数据。

另一种解决方案是在每个广告系列 ID 前加上机器 ID。这会平均分配每台机器之间的行数。这意味着为每个查询创建一个以每个机器 ID 为前缀的查询，但允许增长。

这导致spark。 Spark 将处理在多台机器上运行您的查询并自动为您连接结果，基本上是在执行我上面描述的操作而没有开发开销。

我自己与 Cassandra 合作时，我选择了第一个和第二个解决方案的组合，因为它适合我正在使用的数据结构。请记住，Cassandra 在写入方面非常高效，因此在创建表以帮助过滤查询和更稀疏地存储数据时不要过于保守。

也许通过以日期为前缀的广告系列 ID 的哈希值来存储点击次数对您有用。 编辑：除非禁用，否则 Cassandra 将使用 Murmur3 算法自动散列您的主键。

Answer 2

要模拟您对快速读取和分布式权限的要求，请使用下表定义 -

CREATE TABLE clicks_by_campaign(
 camp_id int,
 createdon bigint,
 pub_id int,
 creative_id int,
 event_code int,
//other fields like ip, user agent ,device etc,
PRIMARY KEY((camp_id,createdon),event_code))

这将有助于在分区之间均匀分布数据。这也将解决我们的第二个和第三个查询 -

2. SELECT * FROM clicks WHERE campaign_id=? AND date_time>=? AND date_time <=? 
  Query will be - 
 SELECT * FROM clicks_by_campaign WHERE token(camp_id, createdon) > token(100, '1111111111111') AND token(camp_id, createdon) <= token(100, '22222222222222')


 3. SELECT * FROM clicks WHERE campaign_id=? AND pub_id=? AND  AND date_time>=? AND date_time <=?  AND event_code=?
 The query will be - 
SELECT * FROM clicks_by_campaign WHERE token(camp_id, createdon) > token(100, '1111111111111') AND token(camp_id, createdon) <= token(100, '22222222222222') AND event_code=10

第一个查询 -

1. SELECT * FROM clicks WHERE campaign_id=?

这确实是 cassandra 中的反模式。我会做的是，批量处理活动数据，每小时-每天-每周-每年。再次考虑活动 ID，我们是否必须一次处理所有数据。 'clicks_by_publisher' 也是如此。

编辑 1

Could you elaborate on what you mean by 'token' ? 

Cassandra 使用分区键对行进行分区。在上面的表定义中，我们结合了 camp_id 和 createdon 值（camp_id 和 createdon 就像 RDBMS 中的复合主键一样）来形成一个分区键。 cassandra 分区器计算结合 camp_id 和 createdon 的哈希值，并决定行去哪个分区。要检索同一行，分区程序需要重新计算哈希值。函数 toke() 就是这样做的。

时间戳表示点击事件发生的时间，该值以毫秒为单位。使用 createdon (type long)，将有助于在分区之间均匀分布行。

以插入语句为例

1. INSERT INTO clicks_by_campaign (camp_id,createdon ,....) values 100,1111111111111,......) the calculated hash, lets say 111 (combining values 100,1111111111111 )  -- this will go in partition 1 
2. INSERT INTO clicks_by_campaign (camp_id,createdon ,....) values (100,2222222222222,......) the calculated hash, lets say 222 (combining values 100,2222222222222 )   -- this will go in partition 2

Java 具有将日期转换为毫秒的 API。以毫秒表示的日期可以使用任何时区转换为任何格式。

事实上，您的用例是设计时间序列数据模型的合适人选。