Асинхронное чтение большого стола Кассандры с использованием Scala/Phantom-DSL

Question

У меня проблема с таблицей, содержащей> 800 тыс. Строк. Мне нужно прочитать строки сверху донизу, чтобы обработать их.

Я использую Scala и Phantom для этой цели.

Вот как выглядит мой стол.

CREATE TABLE raw (
    id uuid PRIMARY KEY, 
    b1 text, 
    b2 timestamp, 
    b3 text, 
    b4 text, 
    b5 text 
) WITH bloom_filter_fp_chance = 0.01 
    AND caching = '{"keys":"ALL", "rows_per_partition":"NONE"}' 
    AND comment = '' 
    AND compaction = {'class': 'org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy'} 
    AND compression = {'sstable_compression': 'org.apache.cassandra.io.compress.LZ4Compressor'} 
    AND dclocal_read_repair_chance = 0.1 
    AND default_time_to_live = 0 
    AND gc_grace_seconds = 864000 
    AND max_index_interval = 2048 
    AND memtable_flush_period_in_ms = 0 
    AND min_index_interval = 128 
    AND read_repair_chance = 0.0 
    AND speculative_retry = '99.0PERCENTILE'; 

До сих пор я пытался прочитать таблицу, используя:

def getAllRecords : Future[Seq[row]] = select.fetch 

или более причудливую Play Enumerator и объединить его с Iteratee

def getAllRecords : Enumerator = select.fetchEnumrator 

Ничего это работает, Кажется, что cassandra/driver/my program всегда пытается прочитать все записи заранее, что мне здесь не хватает?

Answer 1

Вы пробовали ознакомиться с кодом в более крупных тестах чтения?

class IterateeBigReadPerformanceTest extends BigTest with ScalaFutures { 

    it should "read the correct number of records found in the table" in { 
    val counter: AtomicLong = new AtomicLong(0) 
    val result = TestDatabase.primitivesJoda.select 
     .fetchEnumerator run Iteratee.forEach { 
     r => counter.incrementAndGet() 
    } 

    result.successful { 
     query => { 
     info(s"done, reading: ${counter.get}") 
     counter.get() shouldEqual 2000000 
     } 
    } 
    } 
} 

Это не то, что будет читать ваши записи заранее. На самом деле у нас есть тесты, чем на запуск более одного часа, чтобы гарантировать достаточные перерывы в GC, отсутствие накладных расходов на GC, давление перменца/металассы остается в пределах и т. Д.

Если что-то действительно изменилось, это только по ошибке, но это все равно должно быть Работа.