Проблема в полном сканировании таблицы в cassandra

Ответ 1

В Кассандре можно сделать полное сканирование таблицы - действительно, это довольно часто для таких вещей, как Spark. Тем не менее, он обычно не "быстро", поэтому он обескуражен, если вы не знаете, почему вы это делаете. Для ваших актуальных вопросов:

1) Если вы используете CQL, вы почти наверняка используете разделитель Murmur3, поэтому ваш минимальный токен -9223372036854775808 (и максимальный токен - 9223372036854775808).

2) Вы используете session.execute(), который будет использовать стандартную согласованность ONE, которая может не возвращать все результаты в вашем кластере, особенно если вы также пишете на ONE, что я подозреваю вас может быть. Поднимите это значение для ВСЕХ и используйте подготовленные инструкции для ускорения разбора CQL:

 public void run() {
     Cluster cluster = getConnection();
     Session session = cluster.connect("db");
     LOGGER.info("Starting ...");
     actualToken = -9223372036854775808;
     boolean run = true;
     int print = 0;

     while ( run ) {
         if (maxTokenReached(actualToken)) {
             LOGGER.info("Max Token Reached!");
             break;
         }
         SimpleStatement stmt = new SimpleStatement(queryBuilder(actualToken));
         stmt.setConsistencyLevel(ConsistencyLevel.ALL);
         ResultSet resultSet = session.execute(stmt);

         Iterator<Row> rows = resultSet.iterator();
         if ( !rows.hasNext()){
             break;
         }

         List<String> rowIds = new ArrayList<String>();

         while (rows.hasNext()) {
             Row row = rows.next();

             Long leadTime = row.getLong("my_column");
             if (myCondition(myCollumn)) {
                 String rowId = row.getString("key");
                 rowIds.add(rowId);
             }

             if (!rows.hasNext()) {
                 Long token = row.getLong("token(rowid)");
                 if (!rowIds.isEmpty()) {
                     LOGGER.info(String.format("Keys found! RowId's: %s ", rowIds));
                 }
             actualToken = nextToken(token);
             }
         }
      }
     LOGGER.info("Done!");
     cluster.shutdown(); 
  }

public boolean maxTokenReached(Long actualToken){
     return actualToken >= maxToken; 
 }

 public String queryBuilder(Long nextRange) {
     return String.format("select token(key), key, my_column from mytable where token(key) >= %s limit 10000;", nextRange.toString()); 
 }

 public Long nextToken(Long token) {
     return token + 1; 
 }

Ответ 2

Я настоятельно рекомендую использовать Spark - даже в автономном приложении (т.е. без кластера). Он позаботится о разделении разделов и обработает их один за другим. Dead легко использовать:

https://github.com/datastax/spark-cassandra-connector

Ответ 3

Это для общего, что вам нужно делать? Или сценарий одного случая? Я согласен, что это не рекомендуемая вещь, которую вы хотите делать на регулярной основе, но у меня также была проблема, когда мне приходилось читать все строки из ColumnFamily, и я полагался на AllRowsReader от клиент Astyanax. Я вижу, что вы используете драйвер Datastax CQL для подключения к вашему кластеру, но если то, что вы ищете, является доказанным, вам может не понравиться проблема с использованием библиотеки Astyanax.

В моем случае я использовал для чтения все ключи строк, а затем у меня было другое задание для взаимодействия с ColumnFamily с ключами, которые я собрал.

import com.netflix.astyanax.Keyspace;
import com.netflix.astyanax.model.ColumnFamily;
import com.netflix.astyanax.model.ConsistencyLevel;
import com.netflix.astyanax.recipes.reader.AllRowsReader;

import java.util.concurrent.CopyOnWriteArrayList;

...        

private final Keyspace keyspace;
private final ColumnFamily<String, byte[]> columnFamily;

public List<String> getAllKeys() throws Exception {

    final List<String> rowKeys = new CopyOnWriteArrayList<>();

    new AllRowsReader.Builder<>(keyspace, columnFamily).withColumnRange(null, null, false, 0)
        .withPartitioner(null) // this will use keyspace partitioner
        .withConsistencyLevel(ConsistencyLevel.CL_ONE).forEachRow(row -> {
        if (row == null) {
            return true;
        }

        String key = row.getKey();

        rowKeys.add(key);

        return true;
    }).build().call();

    return rowKeys;
}

Существуют различные параметры конфигурации для запуска в нескольких потоках и многие другие вещи, например, я сказал, что я просто запускал это один раз в своем коде и работал очень хорошо, я был бы рад помочь, если бы вы столкнулись с проблемами, заставить его работать.

Надеюсь, что это поможет,

Хосе Луис

Ответ 4

Если вам регулярно нужно выполнять полное сканирование таблицы Cassandra, скажем, для аналитики в Spark, я настоятельно рекомендую вам рассмотреть возможность хранения ваших данных с использованием модели данных, оптимизированной для чтения. Вы можете проверить http://github.com/tuplejump/FiloDB для примера настройки, оптимизированной для чтения на Кассандре.