Distributed caching

[universalmind303]

Right now dozer seems to work pretty well as a standalone service for medium sized data, but can really only vertically scale. Are there plans on allowing a customizable cache such as redis or tikv so that you could distribute across multiple nodes?

for example. Right now if i spun up 3 instances of dozer, all 3 would need to individually manage their own ingestion & caching

flowchart LR
    dozer_ingest_1 --> cache_1 --> cache_reader_1
    dozer_ingest_2 --> cache_2 --> cache_reader_2
    dozer_ingest_3 --> cache_3 --> cache_reader_3

Loading

Ideally I'd like to use a single distributed cache so that I can scale horizontally

flowchart LR
   dozer_ingest_1 & dozer_ingest_2 & dozer_ingest_3 ---> distributed_cache
   distributed_cache --> cache_reader_1 & cache_reader_2 & cache_reader_3    

Loading

[chubei]

This is a good idea.

The way we were going to approach this is that, we'll periodically dump the cache to a file, and when a cache reader starts, it first recovers from the most recent dump, and then continues building the cache from realtime data.

Does this approach make sense to you? Do you have other ideas?

[v3g42]

Hey @universalmind303 . Thanks for the question. We are yet to write about this clearly on our docs but here are a few patterns.

1) DDN pattern ( Like a CDN)

flowchart LR
    
    subgraph dp1[Dozer Pipeline 1 - Region X]
        direction LR
        pipeline[SQL Engine] --> writer[Log Writer]
    end

    subgraph dp2[Dozer Pipeline 2 - Region Y]
        direction LR
        pipeline2[SQL Engine] --> writer2[Log Writer]
    end

   dozer_ingest_1 --->| Users | dp1
   dozer_ingest_2  --->| Transactions | dp1
   dozer_ingest_3 --->| Products | dp2

    dp1 ---> zfs
    dp2 ---> zfs
    subgraph zfs[Distributed File System]
        direction RL
        Users
        Transactions
        Products
    end

    zfs -->| Log Reader | cache_reader_1["Users Api"]   
    zfs -->| Log Reader | cache_reader_2["Transactions Api"]  
    zfs -->| Log Reader | cache_reader_3["Products Api"]

    zfs -->| Log Reader | cache_reader_4["Users Api"]     
    zfs -->| Log Reader | cache_reader_5["Products Api"]     
    
    subgraph api[API - Region 1]
        cache_reader_1
        cache_reader_2
        cache_reader_3
    end

    subgraph api2[API - Region 2]
        cache_reader_4
        cache_reader_5
    end

Loading

This is the pattern we are currently deploying our APIs on Dozer Cloud.

• Each cache reader will build the cache independently using a snapshot mechanism.
• API Clients will use a sticky session to avoid inconsistent reads between API calls.
• It can offer planet-scale deployments.
• We are currently testing with OpenZFS for a distributed file system. This offers a much lower cost deployment as compared to a hosted distributed cache.
• As end to end is controlled through Dozer Types, we can easily orchestrate gRPC typed APis and full functional client side support.

This is achievable by running dozer with sub commands

dozer app run
dozer migrate
dozer api run

2) External Cache

This is the solution you recommended. To use an external cache (Redis, Tikv, Aerospike etc). There is an outstanding issue #868 on our backlog and something we will address if there is interest from the community.

Implementation will require the following

• RwCache can be implemented for different external DBs.
• Indexing methods.
• We currently have query and count methods with various filters. These will be supported through secondary index implementations.

flowchart LR
    
    subgraph dp1[Dozer Pipeline 1 - Region X]
        direction LR
        pipeline[SQL Engine] --> writer[Log Writer]
    end

    subgraph dp2[Dozer Pipeline 2 - Region Y]
        direction LR
        pipeline2[SQL Engine] --> writer2[Log Writer]
    end


   dozer_ingest_1 --->| Users | dp1
   dozer_ingest_2  --->| Transactions | dp1
   dozer_ingest_3 --->| Products | dp2

    zfs["File System"]
    dp1 ---> zfs
    dp2 ---> zfs
    

    subgraph dcache[Distributed Cache]
        direction RL
        Users
        Transactions
        Products
    end

    zfs -->| Log Reader 1 | dcache
    zfs -->| Log Reader 2 | dcache
    zfs -->| Log Reader 3 | dcache

    
    dcache --> cache_reader_1["Users Api"]   
    dcache --> cache_reader_2["Transactions Api"]  
    dcache --> cache_reader_3["Products Api"]
    
    dcache --> cache_reader_4["Users Api"]     
    dcache --> cache_reader_5["Products Api"]     
    
    subgraph api[API - Region 1]
        cache_reader_1
        cache_reader_2
        cache_reader_3
    end

    subgraph api2[API - Region 2]
        cache_reader_4
        cache_reader_5
    end

Loading

3) Custom Implementations

• We expose a log reader that could be used to build any custom functions including writing to an external database.
• We also have bindings for JS and Python.

Eg: Javascript

const dozer = require('..');

async function main() {
    runtime = dozer.Runtime();
    reader = await runtime.create_reader('../.dozer/pipeline', 'trips');

    for (let i = 0; i < 10; ++i) {
        let op = await reader.next_op();
        console.log(op);
    }

    process.exit(0);
}

main();

Eg: Python

import dozer_log

async def main():
    reader = await dozer_log.LogReader.new('../.dozer/pipeline', 'trips')
    for _ in range(10):
        data = await reader.next_op()
        print(data)

if __name__ == '__main__':
    import asyncio
    asyncio.run(main())

This is a relatively new functionality we will write about more in the next release.

[universalmind303]

Thanks for the thorough explanations. I'll take a look at both of these!