🥇NewSQL = NoSQL+ACID

Do niedawna w Odnoklassnikach przechowywano około 50 TB danych przetwarzanych w czasie rzeczywistym w SQL Server. Dla takiej objętości zapewnienie szybkiego, niezawodnego i odpornego na awarie dostępu za pomocą SQL DBMS jest praktycznie niemożliwe. Zazwyczaj w takich przypadkach wykorzystuje się jedną z baz NoSQL, ale nie wszystko można przenieść do NoSQL: niektóre byty wymagają gwarancji transakcji ACID.

To doprowadziło nas do użycia bazy danych NewSQL, czyli DBMS, która zapewnia odporność na awarie, skalowalność i wydajność systemów NoSQL, ale jednocześnie zachowuje klasyczne gwarancje ACID. Liczba działających przemysłowych systemów tej nowej klasy jest niewielka, dlatego sami wdrożyliśmy taki system i uruchomiliśmy go w produkcji.

Jak to działa i co udało się osiągnąć — czytaj dalej.

Dziś miesięczna widownia „Odnoklassników” wynosi ponad 70 milionów unikalnych odwiedzających. My jesteśmy w piątce największych serwisów społecznościowych na świecie i w czołowej dwudziestce stron, na których użytkownicy spędzają najwięcej czasu. Infrastruktura „OK” obsługuje bardzo duże obciążenia: ponad milion żądań HTTP/s na fronty. Części parku serwerów w liczbie ponad 8000 znajdują się blisko siebie — w czterech moskiewskich centrach danych, co pozwala na zapewnienie opóźnienia sieciowego poniżej 1 ms między nimi.

Używamy Cassandry od 2010 roku, zaczynając od wersji 0.6. Dziś w eksploatacji znajduje się kilka dziesiątek klastrów. Najszybszy klaster obsługuje ponad 4 miliony operacji na sekundę, a największy przechowuje 260 TB.

Jednak to wszystko są zwykłe klastry NoSQL, używane do przechowywania słabo spójnych danych. Chcieliśmy jednak zastąpić główne spójne repozytorium, Microsoft SQL Server, które było używane od powstania „Odnoklassników”. Repozytorium składało się z ponad 300 maszyn SQL Server Standard Edition, na których znajdowało się 50 TB danych — bytów biznesowych. Dane te są modyfikowane w ramach transakcji ACID i wymagają wysokiej spójności.

Aby rozdzielić dane po węzłach SQL Server, używaliśmy zarówno partycjonowania pionowego, jak i poziomego. partycjonowanie (sharding). Historycznie używaliśmy prostego schematu sharding danych: każdej encji przypisywano token — funkcję z ID encji. Encje z tym samym tokenem były umieszczane na jednym serwerze SQL. Relacja typu master-detail była realizowana w ten sposób, żeby tokeny głównego i podrzędnego rekordu zawsze się zgadzały i znajdowały się na tym samym serwerze. W sieci społecznościowej niemal wszystkie rekordy są tworzone w imieniu użytkownika — oznacza to, że wszystkie dane użytkownika w obrębie jednego funkcjonalnego podsystemu są przechowywane na jednym serwerze. Innymi słowy, w transakcji biznesowej niemal zawsze brały udział tabele jednego serwera SQL, co pozwalało na zapewnienie spójności danych za pomocą lokalnych transakcji ACID, bez konieczności używania wolnych i niesolidnych rozproszonych transakcji ACID.

Dzięki shardingowi oraz w celu przyspieszenia działania SQL:

Nie używamy ograniczeń Foreign key, ponieważ przy sharding ID encji może znajdować się na innym serwerze.
Nie używamy procedur składowanych i triggerów z powodu dodatkowego obciążenia procesora bazy danych.
Nie używamy JOINów z powodu wszystkich wyżej wymienionych rzeczy i wielu przypadkowych odczytów z dysku.
Poza transakcją, aby zredukować blokady, używamy poziomu izolacji Read Uncommitted.
Wykonujemy tylko krótkie transakcje (średnio krótsze niż 100 ms).
Nie używamy wielowierszowych UPDATE i DELETE z powodu dużej liczby blokad — aktualizujemy tylko po jednym rekordzie.
Zapytania zawsze wykonujemy tylko na indeksach — zapytanie z planem pełnego przeszukiwania tabeli oznacza dla nas przeciążenie bazy danych i jej awarię.

Te kroki pozwoliły wycisnąć z serwerów SQL niemal maksymalną wydajność. Jednak pojawiało się coraz więcej problemów. Przyjrzyjmy się im.

Problemy z SQL

Ponieważ używaliśmy własnego sharding, dodawanie nowych shardów było wykonywane przez administratorów ręcznie. Przez cały ten czas skalowalne repliki danych nie obsługiwały zapytań.
W miarę wzrostu liczby rekordów w tabeli spada prędkość wstawiania i modyfikacji, a przy dodawaniu indeksów do istniejącej tabeli prędkość spada wielokrotnie, tworzenie i odbudowa indeksów odbywa się z przestojem.
Obecność w produkcji niewielkiej liczby Windows dla SQL Server utrudnia zarządzanie infrastrukturą.

Ale głównym problemem jest —

Odporność na awarie

Klasyczny serwer SQL ma słabą odporność na awarie. Załóżmy, że masz tylko jeden serwer baz danych i zawodzi on raz na trzy lata. W tym czasie strona internetowa przestaje działać na 20 minut, co jest do zaakceptowania. Jeśli masz 64 serwery, to strona przestaje działać raz co trzy tygodnie. A jeśli masz 200 serwerów, to strona nie działa co tydzień. To jest problem.

Co można zrobić, aby zwiększyć odporność na awarie serwera SQL? Wikipedia sugeruje nam zbudować wysokodostępny klaster: w którym w przypadku awarii któregokolwiek z komponentów istnieje duplikat.

To wymaga parku kosztownego sprzętu: liczne duplikacje, światłowody, wspólne magazyny, a także włączenie rezerwy działa niepewnie: około 10% włączeń kończy się awarią węzła zapasowego, zależnie od głównego węzła.

Ale główną wadą takiego wysokodostępnego klastra jest zerowa dostępność w przypadku awarii centrum danych, w którym się znajduje. „Odnoklassniki” mają cztery centra danych i musimy zapewnić działanie przy całkowitej awarii jednego z nich.

W tym celu moglibyśmy zastosować replikację Multi-Master , wbudowaną w SQL Server. To rozwiązanie jest znacznie droższe z powodu kosztów oprogramowania i cierpi na dobrze znane problemy z replikacją – nieprzewidywalne opóźnienia transakcji przy replikacji synchronizacyjnej i opóźnienia w stosowaniu replikacji (a w konsekwencji utracone modyfikacje) przy asynchronicznej. Oczekujące ręczne rozwiązywanie konfliktów czyni tę opcję całkowicie niepraktyczną dla nas.

Wszystkie te problemy wymagały radykalnego rozwiązania i przystąpiliśmy do ich szczegółowej analizy. Tutaj musimy zapoznać się z tym, co głównie robi SQL Server – transakcjami.

Prosta transakcja

Rozważmy najprostszą, z punktu widzenia programisty SQL, transakcję: dodanie zdjęcia do albumu. Albumy i zdjęcia są przechowywane w różnych tabelach. Album ma licznik publicznych zdjęć. W takim przypadku transakcja dzieli się na następujące kroki:

Blokujemy album po kluczu.
Tworzymy wpis w tabeli zdjęć.
Jeśli zdjęcie ma publiczny status, zwiększamy w albumie licznik publicznych zdjęć, aktualizujemy wpis i zatwierdzamy transakcję.

Lub w postaci pseudokodu:

TX.start("Albums", id);
Album album = albums.lock(id);
Photo photo = photos.create(…);

if (photo.status == PUBLIC) {
    album.incPublicPhotosCount();
}
album.update();

TX.commit();

Widzę, że najczęstszy scenariusz transakcji biznesowej polega na odczytywaniu danych z bazy danych do pamięci serwera aplikacji, wprowadzaniu pewnych zmian i zapisywaniu nowych wartości z powrotem do bazy danych. Zazwyczaj w takiej transakcji aktualizujemy kilka encji, kilka tabel.

Podczas wykonywania transakcji może dojść do konkurencyjnej modyfikacji tych samych danych z innego systemu. Na przykład, system antyspamowy może uznać, że użytkownik jest podejrzany, a zatem wszystkie zdjęcia użytkownika nie powinny być już publiczne, powinny zostać wysłane do moderacji, co oznacza, że trzeba zmienić photo.status na inną wartość i zaktualizować odpowiednie liczniki. Oczywiście, jeśli ta operacja będzie odbywać się bez gwarancji atomowości i izolacji konkurencyjnych modyfikacji, jak w ACID, to rezultat nie będzie tym, czego potrzebujemy — albo licznik zdjęć pokaże błędną wartość, albo nie wszystkie zdjęcia zostaną wysłane do moderacji.

Takiego kodu, manipulującego różnymi encjami biznesowymi w ramach jednej transakcji, napisano w całej historii portalu Odnoklassniki bardzo dużo. Według doświadczeń migracji na NoSQL z Spóźniona spójność wiemy, że największe trudności (i wydatki czasowe) wynikają z konieczności opracowywania kodu zapobiegającego naruszeniom spójności danych. Dlatego głównym wymaganiem wobec nowego magazynu uznaliśmy zapewnienie dla logiki aplikacyjnej prawdziwych transakcji ACID.

Inne, nie mniej ważne wymagania to:

W przypadku awarii centrum danych zarówno odczyt, jak i zapis do nowego magazynu powinny być dostępne.
Zachowanie obecnej prędkości rozwoju. To znaczy, podczas pracy z nowym magazynem ilość kodu powinna być mniej więcej taka sama, nie powinno być potrzeby dopisywania czegokolwiek do magazynu, opracowywania algorytmów rozwiązywania konfliktów, utrzymywania wtórnych indeksów itp.
Prędkość działania nowego magazynu powinna być wystarczająco wysoka zarówno podczas odczytu danych, jak i przy przetwarzaniu transakcji, co w praktyce oznaczało niewłaściwość akademicko rygorystycznych, uniwersalnych, ale wolnych rozwiązań, jak na przykład dwuetapowych zatwierdzeń.
Automatyczne skalowanie w czasie rzeczywistym.
Wykorzystanie zwykłych, tanich serwerów, bez potrzeby zakupu egzotycznego sprzętu.
Możliwość rozwoju przechowywania przez deweloperów firmy. Innymi słowy, priorytetowo traktowano własne lub oparte na otwartym kodzie rozwiązania, najlepiej w Java.

Rozwiązania, rozwiązania

Analizując możliwe rozwiązania, doszliśmy do dwóch potencjalnych wyborów architektury:

Pierwszy — wziąć dowolny serwer SQL i zaimplementować potrzebną odporność na awarie, mechanizm skalowania, odporny klaster, rozwiązanie konfliktów oraz rozproszone, niezawodne i szybkie transakcje ACID. Oceniamy tę opcję jako dość nietrywialną i czasochłonną.

Druga opcja — wziąć gotowe przechowywanie NoSQL z zaimplementowanym skalowaniem, odpornym klastrem, rozwiązaniem konfliktów i zrealizować transakcje oraz SQL samodzielnie. Na pierwszy rzut oka realizacja SQL, nie mówiąc już o transakcjach ACID, wydaje się zadaniem na lata. Ale potem zrozumieliśmy, że zbiór możliwości SQL, które wykorzystujemy w praktyce, daleki jest od ANSI SQL tak samo, jak Cassandra CQL jest dalekie od ANSI SQL. Przyglądając się bliżej CQL, zrozumieliśmy, że jest wystarczająco bliskie temu, czego potrzebujemy.

Cassandra i CQL

Czym więc interesuje Cassandra, jakie ma możliwości?

Przede wszystkim można tworzyć tabele obsługujące różne typy danych, można wykonać SELECT lub UPDATE po kluczu głównym.

CREATE TABLE photos (id bigint KEY, owner bigint,…);
SELECT * FROM photos WHERE id=?;
UPDATE photos SET … WHERE id=?;

Aby zapewnić spójność danych replik, Cassandra wykorzystuje podejście kworumowe.. W najprostszym przypadku oznacza to, że przy umieszczaniu trzech replik tego samego wiersza na różnych węzłach klastra, zapis jest uznawany za udany, jeśli większość węzłów (tj. dwa z trzech) potwierdziła powodzenie tej operacji zapisu. Dane wiersza są uważane za spójne, jeśli podczas odczytu zostało przeskanowanych większość węzłów, które potwierdziły ich. Tak więc, przy posiadaniu trzech replik, zapewniona jest pełna i natychmiastowa spójność danych w przypadku awarii jednego z węzłów. Takie podejście pozwoliło nam wdrożyć jeszcze bardziej niezawodny schemat: zawsze wysyłać zapytania do wszystkich trzech replik, czekając na odpowiedź od dwóch najszybszych. Opóźniona odpowiedź trzeciej repliki w takim przypadku jest ignorowana. Opoźniony w odpowiedzi węzeł może mieć poważne problemy — przerwy w działaniu, zbieranie pamięci w JVM, reclaim pamięci w jądrze linux, awaria sprzętu, utrata połączenia z siecią. Jednak na operacje klienta i na dane to nie wpływa.

Podejście, w którym zwracamy się do trzech węzłów, a otrzymujemy odpowiedź od dwóch, nazywa się spekulacją: zapytanie do zbędnych replik jest wysyłane jeszcze przed tym, jak „odpadnie”.

Kolejną z zalet Cassandry jest Batchlog — mechanizm, który gwarantuje całkowite zastosowanie lub całkowite niezastosowanie pakietu wprowadzanych przez Ciebie zmian. Pozwala to nam rozwiązać A w ACID — atomowość z pudełka.

Najbliższym do transakcji w Cassandra są tzw. „transakcje lekkie“. Ale od „prawdziwych” transakcji ACID są one dalekie: w rzeczywistości to możliwość przeprowadzenia CAS na danych tylko jednego rekordu, używając konsensusu zgodnie z ciężkim protokołem Paxos. Dlatego prędkość takich transakcji nie jest wysoka.

Czego nam zabrakło w Cassandrze

Tak więc, musieliśmy zaimplementować w Cassandrze prawdziwe transakcje ACID. Dzięki którym moglibyśmy łatwo wdrożyć dwie inne wygodne funkcje klasycznych DBMS: spójne szybkie indeksy, które pozwoliłyby nam na wykonywanie selekcji danych nie tylko wg klucza podstawowego oraz zwykły generator monotonicznych autoinkrementacyjnych ID.

C*One

Tak powstała nowa baza danych C*One, składająca się z trzech typów węzłów serwerowych:

Magazyny — (prawie) standardowe serwery Cassandry, odpowiedzialne za przechowywanie danych na lokalnych dyskach. W miarę wzrostu obciążenia i objętości danych ich liczba może być łatwo skalowana do dziesiątek i setek.
Koordynatory transakcji — zapewniają wykonanie transakcji.
Klienci — serwery aplikacji realizujące operacje biznesowe i inicjujące transakcje. Takich klientów może być tysiące.

Serwery wszystkich typów są częścią wspólnego klastra, używają wewnętrznego protokołu wiadomości Cassandra do komunikacji ze sobą oraz protokół gossip. do wymiany informacji o klastrze. Dzięki Heartbeat serwery dowiadują się o wzajemnych awariach, utrzymują jednolitą strukturę danych — tabele, ich strukturę i replikację; schemat partycjonowania, topologię klastra itp.

Klienci

Zamiast standardowych sterowników używa się trybu Fat Client. Taka noda nie przechowuje danych, ale może pełnić rolę koordynatora wykonania zapytań, czyli Klient sam pełni funkcję koordynatora swoich zapytań: pyta repliki magazynu i rozwiązuje konflikty. To nie tylko bardziej niezawodne i szybsze niż standardowy sterownik, wymagający komunikacji zdalnym koordynatorem, ale także pozwala zarządzać przekazywaniem zapytań. Poza otwartą na kliencie transakcją zapytania są skierowane do magazynów. Jeśli jednak klient otworzył transakcję, to wszystkie zapytania w ramach transakcji są kierowane do koordynatora transakcji.

Koordynator transakcji C*One

Koordynator — to, co wdrożyliśmy dla C*One od podstaw. Odpowiada za zarządzanie transakcjami, blokadami i kolejnością zastosowania transakcji.

Dla każdej obsługiwanej transakcji koordynator generuje znacznik czasu: każdy kolejny jest większy niż w przypadku poprzedniej transakcji. Ponieważ w systemie Cassandra mechanizm rozwiązywania konfliktów opiera się na znacznikach czasu (spośród dwóch konfliktowych rekordów aktualny uznawany jest za ten z późniejszym znacznikiem czasu), konflikt zawsze zostanie rozwiązany na korzyść kolejnej transakcji. W ten sposób zrealizowaliśmy zegar Lamporta — tani sposób na rozwiązywanie konfliktów w systemie rozproszonym.

Blokady

Aby zapewnić izolację, postanowiliśmy użyć najprostszej metody — pesymistycznych blokad na podstawie klucza głównego rekordu. Innymi słowy, w transakcji rekord należy najpierw zablokować, a dopiero potem przeczytać, zmodyfikować i zapisać. Tylko po udanym zatwierdzeniu zapis może zostać odblokowany, aby konkurencyjne transakcje mogły z niego skorzystać.

Realizacja takiej blokady jest prosta w środowisku nienarodowym. W systemie rozproszonym istnieją dwie główne drogi: można albo zrealizować rozproszoną blokadę w klastrze, albo rozdzielić transakcje w taki sposób, aby transakcje dotyczące jednego zapisu zawsze były obsługiwane przez tego samego koordynatora.

Ponieważ w naszym przypadku dane są już rozproszone w grupach lokalnych transakcji w SQL, zdecydowano się przypisać koordynatorom grupy lokalnych transakcji: jeden koordynator wykonuje wszystkie transakcje z tokenem od 0 do 9, drugi — z tokenem od 10 do 19, i tak dalej. W rezultacie każdy z instancji koordynatora staje się mistrzem grupy transakcji.

Wtedy blokady mogą być zrealizowane w postaci prostej HashMap w pamięci koordynatora.

Awaria koordynatorów

Ponieważ jeden koordynator wyłącznie obsługuje grupę transakcji, bardzo ważne jest szybkie określenie faktu jego awarii, aby ponowna próba wykonania transakcji zmieściła się w czasie oczekiwania. Aby było to szybkie i niezawodne, zastosowaliśmy pełnołączny kworumowy protokół heartbeat:

W każdym centrum danych znajduje się co najmniej dwie węzły koordynatora. Okresowo każdy koordynator wysyła wiadomość heartbeat do pozostałych koordynatorów, informując ich o swoim działaniu, a także o tym, z jakich koordynatorów w klastrze otrzymał ostatnio wiadomości heartbeat.

Otrzymując podobne informacje od innych w ramach ich wiadomości heartbeat, każdy koordynator decyduje dla siebie, które węzły klastra działają, a które nie, kierując się zasadą kworum: jeśli węzeł X otrzymał od większości węzłów w klastrze informację o normalnym odbiorze wiadomości z węzła Y, to znaczy, że Y działa. I odwrotnie, jak tylko większość poinformuje o utracie wiadomości z węzła Y, to znaczy, że Y uległ awarii. Ciekawe, że jeśli kworum poinformuje węzeł X, że nie otrzymuje od niego więcej wiadomości, to sama węzeł X uzna siebie za uszkodzoną.

Heartbeat-y są wysyłane z dużą częstotliwością, około 20 razy na sekundę, z okresem 50 ms. W Javie trudno jest zagwarantować odpowiedź aplikacji w ciągu 50 ms z powodu porównywalnego czasu trwania pauz spowodowanych przez zbieracza śmieci. Udało nam się osiągnąć taki czas reakcji, używając zbieracza śmieci G1, który pozwala określić cel dotyczący długości pauz GC. Jednak czasami, dość rzadko, pauzy zbieracza przekraczają 50 ms, co może prowadzić do fałszywego wykrycia awarii. Aby temu zapobiec, koordynator nie zgłasza awarii zdalnego węzła przy utracie pierwszego heartbeat-u od niego, tylko jeśli zginęło kilka pod rząd. W ten sposób udało nam się osiągnąć wykrywanie awarii węzła koordynatora w 200 ms.

Jednak to za mało, aby szybko zrozumieć, który węzeł przestał działać. Trzeba coś z tym zrobić.

Rezerwa

Klasyczny schemat zakłada w przypadku awarii głównego węzła uruchomienie wyborów nowego za pomocą jednego z nowoczesnych uniwersalnych algorytmów. Jednak takie algorytmy mają dobrze znane problemy ze zbieżnością w czasie i długością samego procesu wyboru. Udało nam się uniknąć takich dodatkowych opóźnień dzięki schematowi zastępowania koordynatorów w pełni połączonej sieci:

Załóżmy, że chcemy wykonać transakcję w grupie 50. Z góry zdefiniujmy schemat zastępowania, czyli które węzły będą realizować transakcje grupy 50 w przypadku awarii głównego koordynatora. Naszym celem jest zachowanie funkcjonalności systemu w przypadku awarii datacentrum. Określimy, że pierwszym rezerwowym będzie węzeł z innego datacentrum, a drugim rezerwowym — węzeł z trzeciego. Ten schemat jest wybierany raz i nie zmienia się, dopóki nie zmieni się topologia klastra, co zdarza się bardzo rzadko. Kolejność wyboru nowego aktywnego głównego węzła w przypadku awarii starego będzie zawsze taka: aktywnym głównym węzłem zostanie pierwszy rezerwowy, a jeśli i on przestanie funkcjonować — drugi rezerwowy.

Ten schemat jest niezawodniejszy niż uniwersalny algorytm, ponieważ do aktywacji nowego głównego węzła wystarczy stwierdzenie faktu awarii starego.

Ale jak klienci zrozumieją, który z mistrzów aktualnie pracuje? W ciągu 50 ms niemożliwe jest rozesłanie informacji do tysięcy klientów. Może zdarzyć się sytuacja, w której klient wysyła zapytanie o otwarcie transakcji, nie wiedząc jeszcze, że ten mistrz już nie funkcjonuje, a zapytanie utknie na przekroczeniu limitu czasu. Aby temu zapobiec, klienci spekulatywnie wysyłają zapytanie o otwarcie transakcji jednocześnie do mistrza grupy oraz obu jego rezerw, ale na to zapytanie odpowie tylko ten, kto jest aktywnym mistrzem w danym momencie. Całą późniejszą komunikację w ramach transakcji klient będzie prowadził tylko z aktywnym mistrzem.

Rezerwowi mistrzowie umieszczają otrzymane zapytania dotyczące nie swoich transakcji w kolejce nowo powstałych transakcji, gdzie pozostają przez pewien czas. Jeśli aktywny mistrz umiera, nowy mistrz przetwarza zapytania o otwarcie transakcji ze swojej kolejki i odpowiada klientowi. Jeśli klient już zdążył otworzyć transakcję ze starym mistrzem, to druga odpowiedź jest ignorowana (i oczywiście taka transakcja się nie zakończy i zostanie powtórzona przez klienta).

Jak działa transakcja

Załóżmy, że klient wysłał koordynatorowi zapytanie o otwarcie transakcji dla jakiegoś bytu z danym kluczem głównym. Koordynator blokuje ten byt i umieszcza go w tabeli blokad w pamięci. W razie potrzeby koordynator odczytuje ten byt z magazynu i zapisuje uzyskane dane w stanie transakcji w pamięci koordynatora.

Gdy klient chce zmienić dane w transakcji, wysyła do koordynatora zapytanie o modyfikację bytu, a ten umieszcza nowe dane w tabeli stanu transakcji w pamięci. Na tym zapis jest zakończony — zapis w magazynie nie jest dokonywany.

Gdy klient żąda w ramach aktywnej transakcji swoich zmienionych danych, koordynator postępuje następująco:

jeśli ID jest już w transakcji, to dane są pobierane z pamięci;
jeśli ID nie ma w pamięci, to brakujące dane są odczytywane z węzłów magazynowych, łączone z tymi, które już są w pamięci, a wynik przekazywany klientowi.

W ten sposób klient może odczytać własne zmiany, a inni klienci tych zmian nie widzą, ponieważ są one przechowywane tylko w pamięci koordynatora, w węzłach Cassandra jeszcze ich nie ma.

Kiedy klient przesyła commit, stan, który istniał w pamięci serwisu, jest zapisywany przez koordynatora w logged batch, a następnie w tej formie logged batch wysyłany jest do magazynów Cassandra. Magazyny podejmują wszystkie niezbędne działania, aby ten pakiet został atomowo (w całości) zastosowany, a następnie zwracają odpowiedź do koordynatora, który zwalnia blokady i potwierdza pomyślność transakcji klientowi.

Aby cofnąć operację, koordynatorowi wystarczy jedynie zwolnić pamięć zajętą przez stan transakcji.

W wyniku powyżej opisanych usprawnień zrealizowaliśmy zasady ACID:

Atomowość. To gwarancja, że żadna transakcja nie będzie częściowo zarejestrowana w systemie, zostaną wykonane albo wszystkie jej podoperacje, albo żadna. U nas ten zasad przestrzegany jest dzięki logged batch w Cassandra.
Spójność. Każda udana transakcja z definicji rejestruje tylko dopuszczalne wyniki. Jeśli po otwarciu transakcji i wykonaniu części operacji okaże się, że wynik jest niedopuszczalny, następuje cofnąć.
Izolacja. Podczas wykonywania transakcji równoległe transakcje nie powinny wpływać na jej wynik. Konkurencyjne transakcje są izolowane przy użyciu pesymistycznych blokad na koordynatorze. Dla odczytów poza transakcją przestrzegana jest zasada izolacji na poziomie Read Committed.
Odporność. Niezależnie od problemów na dolnych poziomach — awaria zasilania, awaria sprzętu — zmiany wprowadzone w pomyślnie zakończonej transakcji powinny pozostać zachowane po wznowieniu funkcjonowania.

Odczyt według indeksów

Weźmy prostą tabelę:

CREATE TABLE photos (
id bigint primary key,
owner bigint,
modified timestamp,
…)

Ma ona ID (klucz główny), właściciela i datę modyfikacji. Musimy wykonać bardzo proste zapytanie — wybrać dane według właściciela z datą modyfikacji „w ciągu ostatnich 24 godzin”.

SELECT *
WHERE owner=?
AND modified>?

Aby takie zapytanie działało szybko, w klasycznej bazie danych SQL należy stworzyć indeks na kolumnach (owner, modified). Możemy to zrobić dość łatwo, ponieważ teraz mamy gwarancje ACID!

Indeksy w C*One

Istnieje tabela źródłowa ze zdjęciami, w której ID rekordu jest kluczem głównym.

Dla indeksu C*One tworzy nową tabelę, która jest kopią tabeli źródłowej. Klucz zgadza się z wyrażeniem indeksowym, do którego dodatkowo wchodzi klucz główny rekordu z tabeli źródłowej:

Teraz zapytanie dotyczące „właściciela za ostatnie 24 godziny” można przepisać jako wybór z innej tabeli:

SELECT * FROM i1_test
WHERE owner=?
AND modified>?

Spójność danych w tabeli źródłowej photos oraz indeksowej i1 jest automatycznie utrzymywana przez koordynatora. Na podstawie samej struktury danych, przy otrzymaniu zmiany, koordynator generuje i zapamiętuje zmianę nie tylko w tabeli głównej, ale także w kopiach. Żadne dodatkowe działania na tabeli indeksu nie są wykonywane, logi nie są odczytywane, blokady nie są stosowane. Oznacza to, że dodawanie indeksów prawie nie zużywa zasobów i praktycznie nie wpływa na szybkość stosowania modyfikacji.

Dzięki ACID udało nam się zaimplementować indeksy „jak w SQL”. Posiadają one spójność, mogą być skalowane, działają szybko, mogą być złożone i wbudowane w język zapytań CQL. W celu wsparcia indeksów nie trzeba wprowadzać zmian w kodzie aplikacyjnym. Wszystko jest proste, jak w SQL. I co najważniejsze, indeksy nie wpływają na szybkość wykonywania modyfikacji w tabeli źródłowej transakcji.

Co uzyskaliśmy

Opracowaliśmy C*One trzy lata temu i uruchomiliśmy w eksploatacji przemysłowej.

Co więc uzyskaliśmy w rezultacie? Spójrzmy na to na przykładzie podsystemu przetwarzania i przechowywania zdjęć, który jest jednym z najważniejszych typów danych w sieci społecznościowej. Chodzi nie o same zdjęcia, lecz o wszelką metainformację. Obecnie w „Odnośnikach” znajduje się około 20 miliardów takich zapisów, system obsługuje 80 tysięcy zapytań na sekundę, do 8 tysięcy transakcji ACID na sekundę związanych z modyfikacją danych.

Gdy używaliśmy SQL z współczynnikiem replikacji = 1 (ale w RAID 10), metainformacja zdjęć była przechowywana w wysoko dostępnej klastrze z 32 maszyn z Microsoft SQL Server (plus 11 rezerwowych). Dodatkowo wydzielono 10 serwerów do przechowywania kopii zapasowych. Łącznie 50 kosztownych maszyn. System działał przy nominalnym obciążeniu, bez zapasu.

Po migracji do nowego systemu uzyskaliśmy współczynnik replikacji = 3 — po jednej kopii w każdym centrun danych. System składa się z 63 węzłów przechowującego Cassandra oraz 6 maszyn koordynacyjnych, co daje łącznie 69 serwerów. Te maszyny są jednak znacznie tańsze, ich całkowity koszt to około 30% kosztu systemu na SQL. Przy tym obciążenie utrzymuje się na poziomie 30%.

Wprowadzenie C*One znacznie zredukowało opóźnienia: operacja zapisu w SQL zajmowała około 4,5 ms. W C*One — około 1,6 ms. Czas trwania transakcji wynosi średnio mniej niż 40 ms, komit wykonywany jest w 2 ms, a czas odczytu i zapisu wynosi średnio 2 ms. 99. percentyl — zaledwie 3-3,1 ms, liczba timeoutów zmniejszyła się 100 razy — wszystko to dzięki szerokiemu zastosowaniu spekulacji.

Na dzień dzisiejszy z eksploatacji wycofano większość węzłów SQL Server, a nowe produkty są opracowywane wyłącznie z wykorzystaniem C*One. Dostosowaliśmy C*One do pracy w naszej chmurze. one-cloud, co pozwoliło przyspieszyć wdrażanie nowych klastrów, uprościć konfigurację i zautomatyzować eksploatację. Bez dostępu do kodu źródłowego byłoby to znacznie trudniejsze i mniej stabilne.

Obecnie pracujemy nad przeniesieniem innych naszych magazynów do chmury — ale to już zupełnie inna historia.

Źródło: habr.com