Strojenie wydajności Cassandry

Od 8 tys. do 29 tys. zapisów na sekundę

• Definicja keyspace'u i tabeli w Cassandrze
• Konfiguracja węzła Cassandry
• Pomiar przepustowości
• Wydajność ograniczona przez klienta
• Odczyty a zapisy
• Więcej węzłów i więcej dysków

Wzięliśmy bazę IP2Location w wersji DB11. Zawiera kilka milionów zakresów IPv4, które po rozwinięciu dają ponad 2 miliardy adresów. Taka liczba wpisów to w zasadzie nic wielkiego dla RDBMS PostgreSQL czy rozproszonego systemu baz danych Apache Cassandra. Problemem jest jednak szybkość ładowania danych. Pytanie brzmi, jak szybko jestem w stanie programowo wyliczyć adresy IP dla zakresów IP i wstawić je do trwałego magazynu danych.

PostgreSQL bez trudu pomieści około 10 TB danych na jednym węźle. Pomieści nawet więcej, zwłaszcza jeśli dane zostaną podzielone na osobne partycje/tabele albo użyjemy wielu tablespace'ów na osobnych dyskach. O Apache Cassandrze mówi się, że również może przechowywać ponad 10 TB na węzeł. Sprawdźmy to więc!

Definicja keyspace'u i tabeli w Cassandrze #

Zaczynamy od zdefiniowania podstawowego keyspace'u bez redundancji. Bez redundancji, bo najpierw chcemy poznać układ struktury danych.

create keyspace ip2location
with replication = {
  'class': 'SimpleStrategy',
  'replication_factor': 1
};

Definicja tabeli z tylko jednym kluczem partycji.

create table hosts_by_country_and_ip
(
  country varchar,
  ip varchar,
  region varchar,
  city varchar,
  lati varchar,
  longi varchar,
  izp varchar,
  timezone varchar,
  primary key ((country), ip)
);

Przy jednym kluczu partycji (country) i 6 węzłach Cassandry (z domyślnymi ustawieniami z kontenera Dockera) uzyskujemy 8–9 tys. insertów na sekundę. To znacznie poniżej oczekiwanych 20–40 tys. insertów na nowoczesnym sprzęcie. Środowisko testowe nie działa jednak na nowoczesnym sprzęcie – w rzeczywistości jest to HP Z800.

create table hosts_by_country_and_ip
(
  country varchar,
  ip varchar,
  region varchar,
  city varchar,
  lati varchar,
  longi varchar,
  izp varchar,
  timezone varchar,
  primary key ((country, region, city), ip)
);

Teraz inne podejście do struktury tabeli. Zamiast jednego klucza partycji mamy teraz 3 klucze partycji (country, region, city). Cassandra wylicza klucz partycji funkcją token, która daje wynik liczbowy. Domyślnie każdy węzeł Cassandry przechowuje 16 tokenów (a ściślej zakresów wartości tokenów). Przy każdym insercie wyliczana jest wartość tokena, a koordynator kieruje insert do odpowiedniego węzła, który przechowuje ten sam token – czyli dane powiązane z tym tokenem. Po takiej zmianie struktury uzyskujemy 16 tys. insertów na sekundę.

Zapytasz, dlaczego tak jest? W gruncie rzeczy zależy to od wstawianych danych. Nasz program ładujący w Pythonie czyta zakres IPv4 z PostgreSQL i generuje listę adresów IP do wstawienia do Cassandry. Nie użyliśmy żadnej klauzuli ORDER, więc rozkład danych może, ale nie musi być „losowy”. W efekcie koordynator zapisów Cassandry musi rozwiązywać zbyt wiele konfliktów/blokad, bo wiele procesów wstawia dane z tym samym kluczem partycji. Jeśli rozproszymy dane, minimalizujemy czas obliczeniowy potrzebny na rozwiązywanie konfliktów.

Konfiguracja węzła Cassandry #

Najpierw próbowaliśmy dostosować strukturę tabeli, aby zwiększyć wydajność zapisów. Kolejną rzeczą, którą możemy spróbować, jest inny sprzęt i zmienione parametry uruchomieniowe. Zamiast HP Z800 (z dyskami SATA SSD) uruchamiamy więc jeden węzeł Cassandry na maszynie z jednym dyskiem NVMe. Utworzyliśmy ten sam keyspace i tę samą tabelę. Bez zmian. A może jednak była zmiana, bo pojedynczy węzeł okazał się wydajniejszy niż 6 węzłów, ale oczekujemy jeszcze większej przepustowości zapisów. Bez względu na to, czy działało 6, 12 czy 24 instancje ładujące w Pythonie, uzyskujemy do 16 tys. zapisów na sekundę.

Dodatkowo ustawiliśmy jednak następujące wartości konfiguracyjne:

concurrent_reads: 64
concurrent_writes: 128
concurrent_counter_writes: 64

I zwiększyliśmy MAX_HEAP_SIZE z 4 GB do 8 GB. Nie trzymaliśmy się zbytnio dokumentacji, tylko zdrowego rozsądku, zwiększając współbieżność zapisów o arbitralną wartość. Uruchomiliśmy 24 instancje ładujące w Pythonie i w końcu uzyskaliśmy lepsze wyniki – 24 tys. zapisów na sekundę.

Pomiar przepustowości #

Jak w ogóle łatwo zmierzyć przepustowość zapisów? Cassandra nie oferuje żadnej analityki na tym poziomie. Trzeba ręcznie utworzyć liczniki albo użyć Sparka do policzenia rekordów na liście tokenów. Wybraliśmy pierwsze podejście. Są zewnętrzne liczniki, które są aktualizowane przez każdy proces w Pythonie. W ten sposób, w oparciu o logikę boolowską, możemy też realizować wybieranie danych do przetwarzania współbieżnego.

Wydajność ograniczona przez klienta #

Uruchomienie tylko 12 instancji ładujących w Pythonie daje ponad 18 tys. zapisów na sekundę. Istnieje więc związek między możliwościami serwera a liczbą działających zadań wstawiających dane. Zwiększenie liczby procesów do 48 instancji daje 28 tys. zapisów na sekundę. Przy 64 instancjach uzyskaliśmy 29–30 tys. zapisów na sekundę. I myślę, że to maksimum, jakie na tę chwilę da się wycisnąć z tego sprzętu i tej konfiguracji.

Odczyty a zapisy #

Jest jeszcze jedna rzecz do rozważenia w kontekście wydajności. Podczas intensywnych zapisów (np. 12 instancji ładujących) wykorzystanie CPU oscyluje wokół 200–300% (na CPU z 4 rdzeniami i 8 wątkami). Natomiast podczas intensywnych odczytów, przy powiedzmy 32 równoległych procesach, widzimy wykorzystanie CPU między 400 a 700%. To poziom niespotykany podczas zapisów.

...

# Function processing single partition
def process_single_partition(r):
    cluster = Cluster(['0.0.0.0'])
    session = cluster.connect() 
    session.execute("USE ip2location")
    stmt = session.prepare("SELECT ip as c FROM hosts_by_country_and_ip WHERE country=? AND region=? and city=?")
    res = session.execute(stmt, [r[0], r[1], r[2]])   
    counter = 0 
    for row in res:
        counter = counter + 1
    print(r, counter)

# Multiprocessing invocation
Parallel(n_jobs=32)(delayed(process_single_partition)(obj) for obj in records)

Ponieważ zarówno sesje PostgreSQL, jak i Cassandry nie dają się serializować (pickle) przez Parallel, musimy albo użyć puli połączeń, albo po prostu tworzyć połączenie przy każdym uruchomieniu. Na potrzeby tego prostego demo to wystarczy.

Ze względu na rozproszoną naturę Cassandry nie możemy po prostu uruchomić zapytania COUNT. Nawet jeśli użyjemy wszystkich wymaganych kolumn partycji. W tym przykładzie są to country, region i city. Nawet używając ich wszystkich, przy COUNT dostaniemy timeouty, i to nawet wtedy, gdy podczas liczenia nie wykonujemy insertów. Najprawdopodobniej, gdybyśmy wydłużyli odpowiednie parametry timeoutów, uzyskalibyśmy więcej wyników. Używając COUNT, nadal dostaniemy jakieś wyniki, ale nie będzie to rozsądny sposób na faktyczne policzenie liczby naszych rekordów. Właściwym podejściem jest zbudowanie struktur przechowujących liczniki danych albo skorzystanie z diagnostyki Cassandry.

Uwaga: inne możliwości liczenia to DSE Analytics i nodetool tablestats. Interesujący wydaje się też DataStax Bulk Loader.

Więcej węzłów i więcej dysków #

Mówi się, że Cassandra łatwo się skaluje – wystarczy dodawać kolejne węzły, najlepiej z dodatkowymi dyskami. Ogólnie rzecz biorąc to prawda, ale trzeba też brać pod uwagę takie parametry jak rozmiar sterty serwera, dostępna pamięć i wydajność CPU. Jeszcze lepszą wydajność osiągniemy, jeśli backend storage'u zostanie zbudowany z wielu dysków dających więcej IOPS. Warto też pamiętać, że w przeciwieństwie do RDBMS, w Apache Cassandrze (koncepcja Dynamo) struktury musimy projektować pod kątem zapytań, a nie organizacji danych.