🖐🏻 🍲 👨‍👨‍👧‍👦 MariaDBとSphinxを同期するための独自の自転車 🙎 🖇️ 🖍️

2月28日に、私たちのオフィスで開催されたSphinxSearch-meetupでプレゼンテーションを行いました。彼は、フルテキスト検索のためにインデックスを定期的に再構築し、コード内の更新を「インプレース」でレールタイムインデックスに送信し、インデックスとMariaDBデータベースの状態を自動同期する方法について話しました。私のレポートのビデオ録画はリンクから入手できます。ビデオを見るよりも読むことを好む人のために、私はこの記事を書きました。

そもそも、検索がどのようにアレンジされたか、そしてなぜ私たち全員がこれを始めたか。

私たちの検索は、完全に標準的なスキームに従って編成されました。

フロントエンドから、ユーザーリクエストはPHPで記述されたアプリケーションサーバーに送られ、次に彼はデータベースと通信します（MariaDBがあります）。検索を行う必要がある場合、アプリケーションサーバーはバランサー（haproxyがあります）を呼び出します。これは、searchdが実行されているサーバーの1つに接続し、そのサーバーは既に検索を実行して結果を返します。

データベースからのデータは非常に伝統的な方法でインデックスに分類されます。スケジュールに従って、比較的最近更新されたドキュメントで数分ごとにインデックスを再構築し、いわゆる「アーカイブ」ドキュメントでインデックスを再構築します。長い間何も起こりませんでした）。インデックス作成用に割り当てられたマシンがいくつかあります。スクリプトはスケジュールに従って実行され、最初にインデックスを作成し、次に特別な方法でインデックスファイルの名前を変更してから、別のフォルダーに配置します。また、searchdを使用する各サーバーでは、rsyncが1分に1回起動され、このフォルダーからファイルをsearchdインデックスフォルダーにコピーします。その後、何かがコピーされた場合、RELOAD INDEXリクエストを実行します。

ただし、再開と空席の一部の変更では、できるだけ早くインデックスに「到達」する必要がありました。たとえば、パブリックドメインに配置された欠員がパブリケーションから削除された場合、ユーザーの観点から、数秒以内に問題から消えることを期待するのが妥当です。そのため、この種の変更はUPDATEクエリを使用してsearchdに直接送信されます。そして、これらの変更がすべてのサーバー上のインデックスのすべてのコピーに適用されるように、各インデックスに分散インデックスが設定され、すべての検索されたインスタンスに属性の更新が送信されます。アプリケーションサーバーは引き続きバランサーに接続し、分散インデックスを更新する1つのリクエストを送信します。したがって、searchdを使用するサーバーのリストを事前に知る必要も、正確にsearchdを使用するサーバーに到達する必要もありません。

これらはすべてうまくいきましたが、問題がありました。

ドキュメントの作成（これは私たちにとって履歴書または欠員です）とインデックスへのエントリ間の平均遅延は、データベース内のそれらの数に正比例しました。
分散インデックスを使用して属性の更新を送信したため、これらの更新がインデックスのすべてのコピーに適用される保証はありませんでした。
インデックスの再構築中に発生した「緊急」の変更は、 RELOAD INDEX

コマンドの実行時に失われ（単に新しく作成されたインデックスにないため）、次のインデックス再作成後にのみインデックスに反映されます。
searchdを使用してサーバー上のインデックスを更新するためのスクリプトは、互いに独立して実行され、それらの間の同期はありませんでした。このため、異なるサーバーでインデックスを更新する間の遅延は数分に達する可能性があります。
検索に関連する何かをテストする必要がある場合は、変更のたびにインデックスを再構築する必要がありました。

これらの問題のそれぞれは、検索インフラストラクチャを根本的に作り直す価値はありませんでしたが、それらをまとめると、かなり目に見えて人生を台無しにしました。

リアルタイムSphinxインデックスを使用して、上記の問題に対処することにしました。さらに、RTインデックスへの移行だけでは不十分でした。データ競合を完全に取り除くには、アプリケーションからインデックスへのすべての更新が同じチャネルを経由するようにする必要がありました。さらに、インデックスの再構築中にデータベースに加えられた変更をどこかに保存する必要がありました（結局、インデックスを再構築しなければならないことがありますが、手順は瞬時ではありません）。

データ転送チャネルなどのMySQLレプリケーションプロトコルを使用して接続することにしました。MySQLbinlogは、インデックスの再構築中に変更を保存する場所です。このソリューションにより、アプリケーションコードからSphinxに書き込む必要がなくなりました。また、グローバルトランザクションIDで行ベースのレプリケーションを既に使用しているため、データベースレプリカの切り替えは非常に簡単に行えます。

もちろん、そこからインデックスに送信するための変更を取得するためにデータベースに直接接続するという考えは新しいものではありません：2016年に、Avitoの同僚がプレゼンテーションを行い、 Sphinxのデータをメインデータベースと同期する問題をどのように解決したかを詳しく説明しました。 PostgreSQLではなくMariaDB、および古いSphinxブランチ（つまり、バージョン2.3.2）が異なる点を除き、彼らの経験を活用して同様のシステムを自分たちで作成することにしました。

MariaDBの変更をサブスクライブし、Sphinxのインデックスを更新するサービスを作成しました。彼の責任は次のとおりです。

レプリケーションプロトコルを介してMariaDBサーバーに接続し、binlogからイベントを受信します。
現在のバイナリログの位置と最後に完了したトランザクションの番号を追跡します。
binlogイベントのフィルタリング。
インデックスで追加、削除、または更新する必要があるドキュメント、および更新されたドキュメントについて確認する-どのフィールドを更新する必要があるか
MariaDBからの欠落データの要求。
インデックス更新リクエストの生成と実行。
必要に応じてインデックスを再構築します。

go-mysqlライブラリを使用して、レプリケーションプロトコルを使用して接続を確立しました。彼女はMariaDBとの接続を確立し、複製イベントを読み取り、それらをハンドラーに渡す責任があります。このハンドラーは、ライブラリによって制御されるgorutinで始まりますが、ハンドラーコードは独自に記述します。ハンドラーコードでは、イベントが興味のあるテーブルのリストで検証され、これらのテーブルへの変更が処理のために送信されます。ハンドラーはトランザクションステータスも保存します。これは、レプリケーションプロトコルのイベントが順番に並んでいるためです。GTID（トランザクションの開始）-> ROW（データの変更）-> XID（トランザクションの終了）で、最初のトランザクションのみにトランザクション番号に関する情報が含まれています。変更が適用されたバイナリログ内の位置に関する情報を保存するために、トランザクション番号を完了とともに転送する方が便利です。このため、現在のトランザクションの開始から完了までの番号を覚えておく必要があります。

 MySQL [(none)]> describe sync_state; +-----------------+--------+ | Field | Type | +-----------------+--------+ | id | bigint | | dummy_field | field | | binlog_position | uint | | binlog_name | string | | gtid | string | | flavor | string | +-----------------+--------+

searchdを使用して、各サーバー上の1つのドキュメントの特別なインデックスに最後に完了したトランザクションの番号を保存します。サービスの開始時に、インデックスが初期化され、予想される構造になっていること、およびすべてのサーバーに保存された位置が存在し、すべてのサーバーに同じであることを確認します。次に、これらのチェックが成功し、保存された位置からバイナリログの読み取りを開始できた場合、同期手順を開始します。チェックに失敗した場合、または保存された位置からバイナリログの読み取りを開始できなかった場合、保存された位置をMariaDBサーバーの現在の位置にリセットし、インデックスを再構築します。

レプリケーションイベントの処理は、データベースの特定の変更の影響を受けるドキュメントを特定することから始まります。これを行うには、サービスの構成で、対象のテーブルの行変更イベントのルーティング、つまりデータベースの変更のインデックス付け方法を決定するための一連のルールなどを行いました。

 [[ingest]] table = "vacancy" id_field = "id" index = "vacancy" [ingest.column_map] user_id = ["user_id"] edited_at = ["date_edited"] profession = ["profession"] latitude = ["latitude_deg", "latitude_rad"] longitude = ["longitude_deg", "longitude_rad"] [[ingest]] table = "vacancy_language" id_field = "vacancy_id" index = "vacancy" [ingest.column_map] language_id = ["languages"] level = ["languages"] [[ingest]] table = "vacancy_metro_station" id_field = "vacancy_id" index = "vacancy" [ingest.column_map] metro_station_id = ["metro"]

たとえば、このルールセットでは、 vacancy

、 vacancy_language

およびvacancy_metro_station

への変更は、 vacancy

インデックスに含める必要があります。文書番号は、 vacancy

テーブルのid

フィールドと、他の2つのテーブルのvacancy_id

フィールドでvacancy_id

ます。 column_map

フィールドは、異なるデータベーステーブルのフィールドに対するインデックスフィールドの依存関係のテーブルです。

さらに、変更の影響を受けるドキュメントのリストを受け取った場合、それらをインデックスで更新する必要がありますが、すぐには行いません。まず、各ドキュメントの変更を蓄積し、このドキュメントの最後の変更から少し時間が経過すると（100ミリ秒）すぐにインデックスに変更を送信します。

多くの場合、異なるテーブルに影響を与えるいくつかのSQLクエリの助けを借りてドキュメントへの単一の論理変更が発生し、時には完全に異なるトランザクションで実行されるため、多くの場合、不必要なインデックス更新を回避するためにこれを行うことにしました。

簡単な例を挙げます。ユーザーが空席を編集したとします。変更を保存するためのコードは、簡単にするために次のように記述されることがよくあります。

 BEGIN; UPDATE vacancy SET edited_at = NOW() WHERE id = 123; DELETE FROM vacancy_language WHERE vacancy_id = 123; INSERT INTO vacancy_language (vacancy_id, language_id, level) VALUES (123, 1, "fluent"), (123, 2, "technical"); DELETE FROM vacancy_metro_station WHERE vacancy_id = 123; INSERT INTO vacancy_metro_station (vacancy_id, metro_station_id) VALUES (123, 55); ... COMMIT;

つまり、最初にすべての古いレコードがリンクテーブルから削除され、次に新しいレコードが挿入されます。同時に、文書内で何も変更されていなくても、これらの削除および挿入に関するエントリがバイナリログに残っています。

必要なものだけを更新するために、次の操作を行いました。変更された行を並べ替えて、各インデックスとドキュメントのペアについて、すべての変更を時系列順に取得できるようにしました。その後、それらを順番に適用して、最終的にどのテーブルのどのフィールドが最終的に変更され、どのフィールドが変更されていないかを判断できます。その後、 column_map

テーブルcolumn_map

使用して、影響を受けるドキュメントごとに更新する必要があるフィールドとインデックス属性のリストcolumn_map

取得できます。さらに、1つのドキュメントに関連するイベントが次々に到着することはありませんが、異なるトランザクションで実行された場合は「異なる」ように見えます。しかし、どのドキュメントが変更されたかを判断する能力については、これは影響しません。

同時に、このアプローチにより、テキストフィールドに変更がなければ、インデックスの属性のみを更新し、Sphinxへの変更の送信を組み合わせることができました。

そのため、インデックスで更新する必要があるドキュメントを見つけることができます。

多くの場合、binlogからのデータはインデックスを更新するリクエストを作成するのに十分ではないため、binlogを読み込んだ同じサーバーから欠落データを取得します。このために、サービスの設定でデータを受信するためのリクエストテンプレートがあります。

 [data_source.vacancy] #               #   -      id     parts = 4 query = """ SELECT vacancy.id AS `:id`, vacancy.profession AS `profession_text:field`, GROUP_CONCAT(DISTINCT vacancy_language.language_id) AS `languages:attr_multi`, GROUP_CONCAT(DISTINCT vacancy_metro_station.metro_station_id) AS `metro:attr_multi` FROM vacancy LEFT JOIN vacancy_language ON vacancy_language.vacancy_id = vacancy.id LEFT JOIN vacancy_metro_station ON vacancy_metro_station.vacancy_id = vacancy.id GROUP BY vacancy.id """

このテンプレートでは、すべてのフィールドが特別なエイリアス[___]:___

マークされています。

欠落データを受信する要求を生成するときと、インデックスを構築するときの両方で使用されます（これについては後で説明します）。

このタイプのリクエストを作成します：

 SELECT vacancy.id AS `id`, vacancy.profession AS `profession_text`, GROUP_CONCAT(DISTINCT vacancy_language.language_id) AS `languages`, GROUP_CONCAT(DISTINCT vacancy_metro_station.metro_station_id) AS `metro` FROM vacancy LEFT JOIN vacancy_language ON vacancy_language.vacancy_id = vacancy.id LEFT JOIN vacancy_metro_station ON vacancy_metro_station.vacancy_id = vacancy.id WHERE vacancy.id IN (< id ,   >) GROUP BY vacancy.id

次に、各ドキュメントについて、それがこのリクエストの結果であるかどうかを確認します。そうでない場合は、メインテーブルから削除されたことを意味するため、インデックスからも削除できます（このドキュメントに対してDELETE

クエリを実行します）。ある場合は、このドキュメントのテキストフィールドを更新する必要があるかどうかを確認します。テキストフィールドを更新する必要がない場合は、このドキュメントに対してUPDATE

クエリを作成し、そうでない場合はREPLACE

ます。

binlogから読み取ったすべての変更を適用しない状況が発生する可能性があるため、ここでは、失敗した場合にbinlogの読み取りを開始できる位置を維持するロジックを複雑にする必要があることに注意してください。

binlogの読み取りを正常に再開するために、次のことを行いました。データベース内の行変更イベントごとに、このイベントが発生した時点で最後に完了したトランザクションのIDを覚えておいてください。変更をSphinxに送信した後、次のように、安全に読み取りを開始できるトランザクション番号を更新します。蓄積されたすべての変更を処理しなかった場合（一部のドキュメントがキューで「追跡」されなかったため）、まだ適用できていない変更に関連するトランザクションから最も早いトランザクションの数を取得します。そして、蓄積されたすべての変更を適用した場合は、最後に完了したトランザクションの番号を取得します。

結果として何が起こったのかは私たちに合っていましたが、もう1つ重要な点がありました。リアルタイムインデックスのパフォーマンスを長期にわたって許容レベルに維持するには、このインデックスの「チャンク」のサイズと数を小さくする必要がありました。これを行うために、Sphinxには、新しいディスクチャンクを作成するFLUSH RAMCHUNK

リクエストと、すべてのディスクチャンクを1つにマージするOPTIMIZE INDEX

リクエストがあります。最初は、定期的に実行するだけだと思っていましたが、それだけです。しかし、残念なことに、バージョン2.3.2では、インデックスの最適化は機能しません（つまり、かなり高い確率でsearchdが低下することが判明しました）。そのため、特にインデックススキームやトークナイザーの設定が変更された場合など、特に必要な場合があるため、インデックスを1日に1回完全に再構築することにしました。

インデックスを再構築する手順は、いくつかの段階で行われます。

インデクサーの構成を生成します

上記のように、サービス構成にはSQLクエリテンプレートがあります。また、インデクサー構成の形成にも使用されます。

また、構成には、インデックスの作成に必要な他の設定（トークン設定、辞書、リソース消費のさまざまな制限）があります。
MariaDBの現在の位置を保存する

この位置から、searchdを使用してすべてのサーバーで新しいインデックスが使用可能になった後、binlogの読み取りを開始します。
インデクサーを開始します

indexer --config tmp.vacancy.indexer.0.conf --all

形式のコマンドindexer --config tmp.vacancy.indexer.0.conf --all

し、完了を待ちます。さらに、インデックスが部分に分割されている場合、すべての部分の構築を並行して開始します。
サーバーにインデックスファイルをロードします

各サーバーへのアップロードも並行して行われますが、すべてのファイルがすべてのサーバーにアップロードされるまで自然に待機します。サービス設定のファイルをダウンロードするには、ファイルをダウンロードするためのコマンドテンプレートのセクションがあります。
```
 [index_uploader] executable = "rsync" arguments = [ "--files-from=-", "--log-file=<<.DataDir>>/rsync.<<.Host>>.log", "--no-relative", "--times", "--delay-updates", ".", "rsync://<<.Host>>/index/vacancy/", ]
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     
```
サーバーごとに、単にホスト変数の名前に置き換えて、結果のコマンドを実行します。ダウンロードにはrsyncを使用しますが、原則として、stdin内のファイルのリストを受け入れ、これらのファイルをsearchdがインデックスファイルを参照することを期待するフォルダーにダウンロードするプログラムまたはスクリプトです。
同期を停止します

binlogの読み取りを停止し、変更の蓄積を担当するゴルーチンを停止します。
古いインデックスを新しいインデックスに置き換えます

searchdを使用するサーバーごとに、順次クエリRELOAD INDEX vacancy_plain

、 TRUNCATE INDEX vacancy_plain

、 ATTACH INDEX vacancy_plain TO vacancy

ます。インデックスがパーツに分割されている場合、各パーツに対してこれらのクエリを順番に実行します。同時に、本番環境にいる場合は、サーバーでこれらのクエリを実行する前に、バランサーを介して負荷を取り除きます（したがって、 TRUNCATE

とATTACH

間のインデックスに対してSELECTクエリを作成しないようにします）最後のATTACH

要求が完了すると、このサーバーに負荷が返されます。
保存した位置から同期を再開する

すべてのリアルタイムインデックスを新しく作成したインデックスに置き換えるとすぐに、binlogから読み取りを再開し、binlogからのイベントを同期します。インデックス作成が始まる前に保存した位置から開始します。

以下に、MariaDBサーバーからのインデックスのラグのグラフの例を示します。

再インデンクス後のバックログ

ここで、再構築後のインデックスの状態は時間内に戻っていますが、これは非常に短時間で発生することがわかります。

ほぼすべての準備が整ったので、リリースの時間です。徐々にやった。まず、いくつかのサーバーにリアルタイムインデックスを流し込みましたが、そのときの残りは同じように機能しました。同時に、「新しい」サーバー上のインデックスの構造は古いものと変わらなかったため、PHPアプリケーションは、リクエストがリアルタイムインデックスで処理されるかプレーンインデックスで処理されるかを心配せずにバランサーに接続できました。

移行更新配布スキーム

以前に説明した属性の更新も古いスキームに従って送信されましたが、すべてのサーバーの分散インデックスは、プレーンインデックスを持つサーバーにのみUPDATEクエリを送信するように構成されていました。さらに、アプリケーションからのUPDATE要求がリアルタイムインデックスを使用してサーバーに到達した場合、この要求は実行されず、古い方法で構成されたサーバーに送信されます。

リリース後、私たちが望んでいたように、データベースの再開または空席の変更と、対応する変更がインデックスに反映されるまでの遅延を大幅に削減することがわかりました。

リアルタイムインデックスに切り替えた後、テストサーバーで変更を行うたびにインデックスを再構築する必要はありませんでした。そのため、比較的安価に検索に参加して、エンドツーエンドの自動テストを作成することが可能になりました。ただし、（データベースに書き込むクライアントの観点から）binlogからの変更を非同期的に処理するため、自動テストに参加しているドキュメントに関する変更がサービスによって処理され、searchdに送信されるまで待機できるようにする必要がありました。

これを行うために、サービスでエンドポイントを作成しました。これはまさにそれを行います。つまり、指定されたトランザクション番号にすべての変更が適用されるまで待機します。これを行うには、データベースに必要な変更を加えた直後に、MariaDB @@gtid_current_pos

リクエストし、それをサービスのエンドポイントに転送します。この時点までにすべてのトランザクションをこの位置に既に適用している場合、サービスはすぐに応答して続行できると答えます。そうでない場合は、変更の適用を担当するゴルーチンで、このGTIDへのサブスクリプションを作成し、それ（またはそれに続くいずれか）が適用されるとすぐに、クライアントが自動テストを続行できるようにします。

PHPコードでは、次のようになります。

 <?php declare(strict_types=1); use GuzzleHttp\ClientInterface; use GuzzleHttp\RequestOptions; use PDO; class RiverClient { private const REQUEST_METHOD = 'post'; /** * @var ClientInterface */ private $httpClient; public function __construct(ClientInterface $httpClient) { $this->httpClient = $httpClient; } public function waitForSync(PDO $mysqlConnection, PDO $sphinxConnection, string $riverAddr): void { $masterGTID = $mysqlConnection->query('SELECT @@gtid_current_pos')->fetchColumn(); $this->httpClient->request( self::REQUEST_METHOD, "http://{$riverAddr}/wait", [RequestOptions::FORM_PARAMS => ['gtid' => $masterGTID]] ); } }