Network data piles up fast!

Network data piles up fast!

• Over 300 C* servers in production
• Over 1200 servers in EC2
• Over 150TB in C*
• About 90TB of SOLR indexes
• 100TB of cold storage data
• 2 PB of PCAP

Network data piles up fast!

• Over 300 C* servers in production
• Over 1200 servers in EC2
• Over 150TB in C*
• About 90TB of SOLR indexes
• 100TB of cold storage data
• 2 PB of PCAP

And we are growing rapidly!

Right sizing our AWS bill

Right sizing our AWS bill

This is all time series data:

• Lots of writes/reads to recent data
• Some reads and very few writes to older data

Right sizing our AWS bill

This is all time series data:

• Lots of writes/reads to recent data
• Some reads and very few writes to older data

So just move all that older, cold data, to cheaper storage.

How hard could that possibly be?

Problem 1:

Data distributed evenly across all these expensive servers

• Using Size Tiered Compaction
• Can't just migrate old SSTables to new cheap servers

Problem 1:

Data distributed evenly across all these expensive servers

• Using Size Tiered Compaction
• Can't just migrate old SSTables to new cheap servers

Solution: use Date Tiered Compaction?

• We update old data regularly
• What SSTables can you safely migrate?

Result:

Problem 1:

Data distributed evenly across all these expensive servers

• Using Size Tiered Compaction
• Can't just migrate old SSTables to new cheap servers

Solution: use Date Tiered Compaction?

• We update old data regularly
• What SSTables can you safely migrate?

Result:

Problem 2: DSE/SOLR re-index takes forever!

We have nodes with up to 300GB of SOLR indexes

• Bootstraping a new node requires re-index after streaming
• Re-index can take a week or more!!!
• At that pace we simply cannot bootstrap new nodes

Problem 2: DSE/SOLR re-index takes forever!

We have nodes with up to 300GB of SOLR indexes

• Bootstraping a new node requires re-index after streaming
• Re-index can take a week or more!!!
• At that pace we simply cannot bootstrap new nodes

Solution:

• Time sharded clusters in application code
• Fanout searches for large time windows
• Assemble results in code (using same algorithm SOLR does)

Problem 2: DSE/SOLR re-index takes forever!

We have nodes with up to 300GB of SOLR indexes

• Bootstraping a new node requires re-index after streaming
• Re-index can take a week or more!!!
• At that pace we simply cannot bootstrap new nodes

Solution:

• Time sharded clusters in application code
• Fanout searches for large time windows
• Assemble results in code (using same algorithm SOLR does)

Result:

What did we gain?

What did we gain?

We can now migrate older timeshards to cheaper servers!

What did we gain?

We can now migrate older timeshards to cheaper servers!

However:

• Cold data servers are still too expensive
• DevOps time suck is massive
• Product wants us to store even more data!

Throwing ideas at the wall

Throwing ideas at the wall

We have this giant data warehouse, let's use that

• Response time is too slow: 10 seconds to pull single record by ID!
• Complex ETL pipeline where latency measured in hours
• Data model is different
• Read only
• Reliability, etc, etc

Throwing ideas at the wall

We have this giant data warehouse, let's use that

• Response time is too slow: 10 seconds to pull single record by ID!
• Complex ETL pipeline where latency measured in hours
• Data model is different
• Read only
• Reliability, etc, etc

What about Elastic Search, HBase, Hive/Parquet, MongoDB, etc, etc?

Throwing ideas at the wall

We have this giant data warehouse, let's use that

• Response time is too slow: 10 seconds to pull single record by ID!
• Complex ETL pipeline where latency measured in hours
• Data model is different
• Read only
• Reliability, etc, etc

What about Elastic Search, HBase, Hive/Parquet, MongoDB, etc, etc?

Wait. Hold on! This Parquet thing is interesting...

Parquet

Parquet

Columnar

• Projections are very efficient
• Enables vectorized execution

Parquet

Columnar

• Projections are very efficient
• Enables vectorized execution

Compressed

• Repeated values become cheap
• Using Run Length Encoding
• Throw Snappy, LZO, or LZ4 on top of that

Parquet

Columnar

• Projections are very efficient
• Enables vectorized execution

Compressed

• Repeated values become cheap
• Using Run Length Encoding
• Throw Snappy, LZO, or LZ4 on top of that

Schema

• Files encode schema and other meta-data
• Support exists for merging disparate schema amongst files

Parquet: Some details

Lots and lots of files

Sizing parquet

Lots and lots of files

Sizing parquet

• Parquet docs recommend 1GB files for HDFS

Lots and lots of files

Sizing parquet

• Parquet docs recommend 1GB files for HDFS
• For S3 the sweet spot appears to be 128 to 256MB

Lots and lots of files

Sizing parquet

• Parquet docs recommend 1GB files for HDFS
• For S3 the sweet spot appears to be 128 to 256MB

We have terabytes of files

Scans take forever!

Lots and lots of files

Sizing parquet

• Parquet docs recommend 1GB files for HDFS
• For S3 the sweet spot appears to be 128 to 256MB

We have terabytes of files

Scans take forever!

Partitioning

Our queries are always filtered by:

1. Customer
2. Time

Partitioning

Our queries are always filtered by:

1. Customer
2. Time

So we Partition by:

├── cid=X
|   ├── year=2015
|   └── year=2016
|        └── month=0
|            └── day=0
|                └── hour=0
└── cid=Y

Partitioning

Our queries are always filtered by:

1. Customer
2. Time

So we Partition by:

├── cid=X
|   ├── year=2015
|   └── year=2016
|        └── month=0
|            └── day=0
|                └── hour=0
└── cid=Y

Spark understands and translates query filters to this folder structure

Queries spanning large time windows

select count(*) from events where ip = '192.168.0.1' and cid = 1 and year = 2016

Queries spanning large time windows

select count(*) from events where ip = '192.168.0.1' and cid = 1 and year = 2016

├── cid=X
|   ├── year=2015
|   └── year=2016
|        |── month=0
|        |   └── day=0
|        |       └── hour=0
|        |           └── 192.168.0.1_was_NOT_here.parquet
|        └── month=1
|            └── day=0
|                └── hour=0
|                    └── 192.168.0.1_WAS_HERE.parquet
└── cid=Y

Problem #1

• Requires listing out all the sub-dirs for large time ranges.
• Remember S3 is not really a file system
• Slow!

Problem #1

• Requires listing out all the sub-dirs for large time ranges.
• Remember S3 is not really a file system
• Slow!

Problem #2

• Pulling potentially thousands of files from S3.
• Slow and Costly!

Solving problem #1

Put partition info and file listings in a db ala Hive.

Solving problem #1

Put partition info and file listings in a db ala Hive.

Why not just use Hive?

Solving problem #1

Put partition info and file listings in a db ala Hive.

Why not just use Hive?

• Still not fast enough
• Also does not help with Problem #2

DSE/SOLR to the Rescue!

DSE/SOLR to the Rescue!

Store file meta data in SOLR

DSE/SOLR to the Rescue!

Store file meta data in SOLR

Efficiently skip elements of partition hierarchy!

select count(*) from events where month = 6

DSE/SOLR to the Rescue!

Store file meta data in SOLR

Efficiently skip elements of partition hierarchy!

select count(*) from events where month = 6

Avoids pulling all meta in Spark driver

1. Get partition counts and schema info from SOLR driver-side
2. Submit SOLR RDD to cluster
3. Run mapPartitions on SOLR RDD and turn into Parquet RDDs

As an optimization for small file sets we pull the SOLR rows driver side

Solving problem #2

Still need to pull potentially thousands of files to answer our query!

Solving problem #2

Still need to pull potentially thousands of files to answer our query!

Can we partition differently?

Solving problem #2

Still need to pull potentially thousands of files to answer our query!

Can we partition differently?

Searching High Cardinality Data

Searching High Cardinality Data

Assumptions

1. Want to reduce # of files pulled for a given query
2. Cannot store all exact values in SOLR
3. We are okay with a few false positives

Searching High Cardinality Data

Assumptions

1. Want to reduce # of files pulled for a given query
2. Cannot store all exact values in SOLR
3. We are okay with a few false positives

This sounds like a job for...

Towards a "Searchable" Bloom Filter

Towards a "Searchable" Bloom Filter

Normal SOLR index looks vaguely like

Towards a "Searchable" Bloom Filter

Normal SOLR index looks vaguely like

Terms are going to grow out of control

Towards a "Searchable" Bloom Filter

Normal SOLR index looks vaguely like

Terms are going to grow out of control

If only we could constrain to a reasonable number of values?

Terms as a "bloom filter"

Terms as a "bloom filter"

Terms as a "bloom filter"

What if our terms were the offsets of the Bloom Filter values?

Terms as a "bloom filter"

What if our terms were the offsets of the Bloom Filter values?

Searchable Bloom Filters

Key Lookups

Key Lookups

Need to retain this C* functionality

Key Lookups

Need to retain this C* functionality

Spark/Parquet has no direct support

What partition would it choose?

The partition would have to be encoded in the key?! 🤔

Key Lookups

Need to retain this C* functionality

Spark/Parquet has no direct support

What partition would it choose?

The partition would have to be encoded in the key?! 🤔

Solution:

• Our keys have time encoded in them
• Enables us to generate the partition path containing the row

Key Lookups

Need to retain this C* functionality

Spark/Parquet has no direct support

What partition would it choose?

The partition would have to be encoded in the key?! 🤔

Solution:

• Our keys have time encoded in them
• Enables us to generate the partition path containing the row

Other reasons to "customize"

Other reasons to "customize"

Parquet has support for filter pushdown

Other reasons to "customize"

Parquet has support for filter pushdown

Spark has support for Parquet filter pushdown, but...

Other reasons to "customize"

Parquet has support for filter pushdown

Spark has support for Parquet filter pushdown, but...

• Uses INT96 for Timestamp
  • No pushdown support: SPARK-11784
  • All our queries involve timestamps!

Other reasons to "customize"

Parquet has support for filter pushdown

Spark has support for Parquet filter pushdown, but...

• Uses INT96 for Timestamp
  • No pushdown support: SPARK-11784
  • All our queries involve timestamps!

• IP Addresses
  • Spark, Impala, Presto have no direct support
  • Use string or binary
  • Wanted to be able to push down CIDR range comparisons

Other reasons to "customize"

Parquet has support for filter pushdown

Spark has support for Parquet filter pushdown, but...

• Uses INT96 for Timestamp
  • No pushdown support: SPARK-11784
  • All our queries involve timestamps!

• IP Addresses
  • Spark, Impala, Presto have no direct support
  • Use string or binary
  • Wanted to be able to push down CIDR range comparisons

Lack of pushdown for these leads to wasted I/O and GC pressure.

Archiving

Archiving

Currently, when Time shard fills up:

1. Roll new hot time shard
2. Run Spark job to Archive data to S3
3. Swap out "warm" shard for cold storage (automagical)
4. Drop the "warm" shard

Archiving

Currently, when Time shard fills up:

1. Roll new hot time shard
2. Run Spark job to Archive data to S3
3. Swap out "warm" shard for cold storage (automagical)
4. Drop the "warm" shard

Not an ideal process, but deals with legacy requirements

Archiving

Currently, when Time shard fills up:

1. Roll new hot time shard
2. Run Spark job to Archive data to S3
3. Swap out "warm" shard for cold storage (automagical)
4. Drop the "warm" shard

Not an ideal process, but deals with legacy requirements

TODO:

1. Stream data straight to cold storage
2. Materialize customer edits in to hot storage
3. Merge hot and cold data at query time (already done)

What have we done so far?

What have we done so far?

1. Time sharded C* clusters with SOLR

What have we done so far?

1. Time sharded C* clusters with SOLR

2. Cheap speedy Cold storage based on S3 and Spark

What have we done so far?

1. Time sharded C* clusters with SOLR

2. Cheap speedy Cold storage based on S3 and Spark

3. A mechanism for archiving data to S3

That's cool, but...

That's cool, but...

How do we handle queries to 3 different stores?

• C*
• SOLR
• Spark

That's cool, but...

How do we handle queries to 3 different stores?

• C*
• SOLR
• Spark

Handle Timesharding and Functional Sharding?

That's cool, but...

How do we handle queries to 3 different stores?

• C*
• SOLR
• Spark

Handle Timesharding and Functional Sharding?

Lot's of Scala query DSL libraries:

• Quill
• Slick
• Phantom
• etc

Lot's of Scala query DSL libraries:

• Quill
• Slick
• Phantom
• etc

1. Simultaneously querying heterogeneous data stores

Lot's of Scala query DSL libraries:

• Quill
• Slick
• Phantom
• etc

1. Simultaneously querying heterogeneous data stores

2. Stitching together time series data from multiple stores

Lot's of Scala query DSL libraries:

• Quill
• Slick
• Phantom
• etc

1. Simultaneously querying heterogeneous data stores

2. Stitching together time series data from multiple stores

3. Managing sharding:

   • Configuration
   • Discovery

Enter Quaero:

Enter Quaero:

Abstracts away data store differences

• Query AST (Algebraic Data Type in Scala)
• Command/Strategy pattern for easily plugging in new data stores

Enter Quaero:

Abstracts away data store differences

• Query AST (Algebraic Data Type in Scala)
• Command/Strategy pattern for easily plugging in new data stores

Deep understanding of sharding patterns

• Handles merging of time/functional sharded data
• Adding shards is configuration driven

Enter Quaero:

Abstracts away data store differences

• Query AST (Algebraic Data Type in Scala)
• Command/Strategy pattern for easily plugging in new data stores

Deep understanding of sharding patterns

• Handles merging of time/functional sharded data
• Adding shards is configuration driven

Exposes a "typesafe" query DSL similar to Phantom or Rogue

• Reduce/eliminate bespoke code for retrieving the same data from all 3 stores