Data Import Handler – sharding

Nasza czytelniczka (pozdrawiamy!) zgłosiła się do nas z problemem dotyczącym współpracy DIH z shardingiem. Wiki projektu SOLR pokazuje moim zdaniem rozwiązanie tej kwestii, ale czyni to trochę na około i przy okazji.

Co to jest sharding?

Sharding oznacza podział danych na kilka części oraz przechowywanie i obróbkę tych danych niezależnie. Dodatkowa logika w ramach aplikacji pozwala na wybranie odpowiedniej części zbioru danych i/lub łączenie wyników z poszczególnych źródeł. W przypadku DIH i shardingu możemy mieć do czynienia z następującym przypadkiem:

• sharding po stronie źródło danych – czyli wiele lokalizacji / tabel zawierających poszczególne części zbioru danych
• sharding po stronie SOLR – czyli podzielenie danych ze źródła na wiele niezależnych instancji SOLR
• oba powyższe jednocześnie

W opisywanym przypadku mamy jeden zbiór danych i chcemy stworzyć wiele zbiorów (tzw. shardów) po stronie SOLR.

Kiedy stosować sharding?

Bardzo ważna kwestia: po co? W moim mniemaniu sharding bywa zbyt często nadużywany generując mnóstwo dodatkowych komplikacji i ograniczeń. Główny powód to duży wolumen danych, które powodują, że indeks SOLR nie mieści się w obrębie jednej maszyny. Jeśli tak nie jest – często oznacza to, że sharding jest zbędny. Kolejny powód to wydajność. Jednak sharding może tutaj pomóc tylko wtedy, gdy inne optymalizacje zawiodą a zapytania są na tyle skomplikowane, że sam narzut shardingu (przekazania zapytania do poszczególnych shardów i łączenie ich odpowiedzi) jest mniejszy niż zysk możliwy do uzyskania.

Dane testowe

Zakładamy jednak, że sharding jest nam potrzebny. W przykładzie poniżej użyłem danych z musicbrainz tworząc prostą tabelę postgresową:

Table "public.albums"

 Column |  Type   |                      Modifiers
--------+---------+-----------------------------------------------------

 id     | integer | not null default nextval('albums_id_seq'::regclass)

 name   | text    | not null

 author | text    | not null

Indexes:

"albums_pk" PRIMARY KEY, btree (id)

W tabeli znajduje się 825661 rekordów. Podkreślam tutaj, że zarówno struktura jak i ilość danych jest na tyle małe, że praktyczna przydatność shardingu jest tu pomijalna.

Instalacja testowa

Do testów użyjemy trzech instancji SOLR. Wszystkie instancje są identyczne, różnica jest związana tylko z numerem portów (8983, 7872, 6761) – testy będą wykonywane na jednej fizycznej maszynie.

Definicja w schema.xml:

<fields>
 <field name="id" type="string" indexed="true" stored="true" required="true" />
 <field name="album" type="text" indexed="true" stored="true" multiValued="true"/>
 <field name="author" type="text" indexed="true" stored="true" multiValued="true"/>
</fields>
<uniqueKey>id</uniqueKey>
<defaultSearchField>album</defaultSearchField>

Definicja DIH w solrconfig.xml:

<requestHandler name="/dataimport" class="org.apache.solr.handler.dataimport.DataImportHandler">
 <lst name="defaults">
  <str name="config">db-data-config.xml</str>
 </lst>
</requestHandler>

I plik DIH db-data-config.xml:

<dataConfig>
 <dataSource driver="org.postgresql.Driver" url="jdbc:postgresql://localhost:5432/shardtest" user="solr" password="secret" />
 <document>
  <entity name="album" query="SELECT * from albums">
   <field column="id" name="id" />
   <field column="name" name="album" />
   <field column="author" name="author" />
  </entity>
 </document>
</dataConfig>

W tym momencie każda instancja jest w stanie dokonać pełnego importu danych.

Zestawiamy sharding

Naszym celem jest takie zmodyfikowanie konfiguracji DIH by każda instancja indeksowała tylko „swoją” część danych. Najprościej zrobić to modyfikując zapytanie pobierające dane np w ten sposób:

SELECT * from albums where id % LICZBA_INSTANCJI = NUMER_INSTANCJI

gdzie:

• LICZBA_INSTANCJI – liczba serwerów SOLR przechowujących unikalne części zbioru danych
• NUMER_INSTANCJI – numer instancji (liczony od zera)

takie zapytanie nie gwarantuje nam dokładnie i idealnie równego podziału ale spełnia dwa konieczne warunki:

• dany rekord trafi zawsze na konkretną i zawszę tę samą instancję
• pojedynczy rekord trafi zawsze na tylko jedną instancję

czyli db-data-config.xml na każdej maszynie różni się teraz zapytaniem i wygląda na poszczególnych instancjach następująco:

• SELECT * from albums where id % 3 = 0
• SELECT * from albums where id % 3 = 1
• SELECT * from albums where id % 3 = 2

Sprawdzamy działanie

Po uruchomieniu wszystkich instancji SOLR na każdej wywołujemy adres:

/solr/dataimport?command=full-import

Po zakończeniu pracy DIH i wywołaniu:

/solr/dataimport?command=status

dostajemy w odpowiedzi od instancji odpowiednio:

• Added/Updated: 275220 documents.
• Added/Updated: 275221 documents.
• Added/Updated: 275220 documents.

Wykonując prostą operację dodawania widzimy, że we wszystkich instancjach łącznie mamy 825661 dokumentów, czyli tyle ile powinno tam być 🙂
Wykonajmy jeszcze zapytanie o wszystkie dokumenty, z wykorzystaniem shardingu wywołując na dowolnej instancji:

/solr/select/?q=*:*&shards=localhost:6761/solr,localhost:7872/solr,localhost:8983/solr

Wynik: 825661.

To działa! 🙂