Я пытаюсь загрузить большой объем данных в кольцо Cassandra с 10 узлами.
Сценарий, выполняющий вставки, получает около 4000 вставок в секунду, предположительно заблокированных на сетевом вводе-выводе. Я запускаю 8 из них на одной машине, и пропускная способность увеличивается почти линейно. (Индивидуальная производительность немного снижается, но это более чем компенсируется дополнительными процессами.)
Это работает прилично, однако мне все еще не хватает пропускной способности, поэтому я запустил ту же настройку еще на 3 ВМ. (Таким образом, 8 процессов * 4 ВМ) После первой дополнительной ВМ и с увеличением частоты и серьезности по мере добавления дополнительных ВМ происходит следующее:
- Клиенты начинают получать ошибки тайм-аута. Они могут повторить попытку записи, но, поскольку они делают это партиями, их дальнейшее продвижение почти полностью исключается.
- Кольцо становится нестабильным, и узлы начинают маркировать себя как «неработающие». Кроме того, разные узлы, как правило, имеют разные представления о том, кто не работает. Кольцо не восстанавливается после прерывания сценариев. (Я даже не смог исправить это, просто перезапустив отдельные узлы: мне пришлось перезапустить все кольцо.)
«Вниз» варьируется. В моем последнем прогоне:
- 4 узла умерли полностью. (Кассандра вообще не работала.) При проверке журналов не было обнаружено ничего, что могло бы объяснить причину ее смерти.
- На пятом бежала Кассандра.
nodetool statusна этом узле висит. Кажется, что два потока находятся в бесконечных циклах. (Они стабильно используют 100% ЦП.) В журналах естьjava.lang.OutOfMemoryError: Java heap space.
Код по существу:
def prepped_batch_insert(session, items, insert_query, silent=False):
# A mapping of number of inserts -> a prepared query for that number of
# inserts.
prepped_statements = {}
def get_prepped_statement(inserts):
if inserts in prepped:
# We already created a prepared query for this many inserts, use
# it:
return prepped_statements[inserts]
else:
# We haven't yet created a prepared query for this many inserts, so
# do so now:
query = ['BEGIN UNLOGGED BATCH']
for idx in xrange(inserts):
query.append(insert_query.query)
query.append('APPLY BATCH;')
query = '\n'.join(query)
ps = session.prepare(query)
prepped_statements[inserts] = ps
return ps
def do_prepped_batch_insert(batch)
ps = get_prepped_statement(len(batch))
# Generate the list of params to the prepared query:
params = []
for idx, item in enumerate(batch):
for k in insert_query.keyorder:
params.append(item[k])
# Do it.
session.execute(ps, params)
return inserter.insert_and_time(
items, # data generator
do_prepped_batch_insert, # The above function
_WHAT_APPEARS_TO_BE_THE_OPTIMAL_CASSANDRA_BATCH_SIZE, # = 200
silent=silent,
)
Функция insert_and_time разбивает items на пакеты по 200 штук, вызывает указанную выше функцию и измеряет время для всего комплекта и kaboodle. (Этот код токсичен для кольца.)
Мы предприняли больше чтений, потому что (мне сказали) 20 000 вставок в секунду было медленным (потребуется некоторое время, чтобы вставить данные, которые я хотел бы вставить с такой скоростью…), и что Cassandra способна работать с высокой пропускной способностью.
Мои вопросы:
- Есть ли что-то необычное в том, что я делаю? Ничего плохого?
- Я просто DDoS-атак на кольцо?
- Как я могу отладить, что не так?
- Ошибочный клиент, ИМХО, никогда не должен убить сервер. (И вышеизложенное не так уж ошибочно.) Что я могу сделать, чтобы предотвратить это?
¹Похоже, что клиент также медленно пропускает файловые дескрипторы. Я не думаю, что это связано. (Я вызываю .shutdown как для кластера, так и для соединения.) Глядя на источник драйвера, кажется, что существует множество путей, где исключение может вызвать утечку.
