こちらはバイセルテクノロジーズ Advent Calendar 2021 の 7日目の記事です。

前日の記事は @naoto_pqさんの「スクラムに慣れてきたら考えるべきこと)」でした。

今日は開発2部の浅香が執筆します。

普段業務ではリユースプラットフォーム内の商品マスタサービスを開発するチームに所属しています。

その中で商品検索用のElasticsearchをインフラからAPI、データ連携までの実装を一通り開発してきたのでその開発手法や選定理由などを紹介しようと思います。

Elasticsearchとは
Elasticsearchを導入した背景
アーキテクチャ選定
helm-chartsでの構築
データ連携について
結び

Elasticsearchとは

Elasticsearchは分散型の検索エンジンです。

データをシャードと呼ばれる単位に分散して保持することによって検索の並列化を実現し、高速検索が可能です。

またレプリカシャードと呼ばれるデータの複製も自動的に作られ、動作しているシャードが障害で使えなくなった際には自動的に切り替えてくれるなど可用性が高いことも特徴の一つです。

Elasticsearchを導入した背景

導入の背景としては

全文検索、曖昧検索が必要なこと
検索レスポンスが高速なこと

の2点が要件としてあったことです。

バイセルでは買取事業にて一日に数千単位の商品を査定員が査定しています。

査定の際にシステムとして画像や一部の情報から商品情報を特定することがありますが、上記2点はSQLを用いての検索では要件を満たせないと判断しElasticsearchを使うことになりました。

アーキテクチャ選定

Elasticsearchはいくつかの構築方法があり、その選択肢をご紹介します。

自分が検討したのは4パターンあります。

Elastic Cloud

Elastic社が提供するクラウドサービスです。

開発元が提供しているために常に最新のElasticsearchが使えることと、環境構築が容易なことがメリットとして挙げられます。

月額契約となるため、サービスの規模に対してコストが見合わないことがデメリットとして挙げられました。

Elastic Cloud on Kubernetes

Elastic社が提供するOSSです。

自身で構築したKubernetes環境にインストールすることによって使用することができます。

基本機能は無料で使用することができ、有償機能でElastic Cloudで提供しているようなサービスを利用することができます。

各クラスタの構成をマニュフェストファイルで書くことができるため、構成の柔軟性を取りやすいことがメリットとして挙げられます。

helm-charts

Elastic社が提供する公式のchartになります。

chartと呼ばれる設定ファイル群をインストールし、コマンド一発で自身で構築したKubernetes環境にデプロイすることができます。設定が容易なことに加えて、YAML形式で変数を指定することである程度柔軟な構成にすることも可能です。

オフィシャルのDockerイメージでコンテナベースで構築

Elastic社が提供するimageを元にコンテナを構築して開発していく方法です。手元のローカル環境での再現に使用しました。

helm-chartsでの構築

この中から今回はhelm-chartsを使用した構築方法を選択しました。

以下の2点が主な理由です。

環境の構築が容易なこと
可用性を考えた時にKubernetesで構築できること

環境構築方法については提供元のREADME.mdを参考にすることで簡単に行うことができました。

# chartをインストール
helm repo add elastic https://helm.elastic.co

# Kubernetesにインストール
Helm 3: helm install elasticsearch elastic/elasticsearch
Helm 2: helm install --name elasticsearch elastic/elasticsearch

データ連携について

データ連携については、Google Cloud BigQueryからElasticsearchへ連携バッチ処理で行う形にしました。

GCPを使用しているためApp Engineで登録処理を実装しました。

GCPのサーバーレス関数といえばCloud Functionsを思い浮かべる方が多くいらっしゃると思いますが、ここでApp Engineを採用した理由としてはタイムアウトの長さが挙げられます。

Cloud Functionsは最大9分なのに対して、App Engineは最大60 分まで設定が可能です。

処理にはGoを採用しました。

処理内容としてはすでにドキュメントが保存されていればupdate、なければinsertするいわゆるupsertの処理を考えていました。

それに対応する関数はないかと探していたところBulk APIを発見しました。説明は以下となっており、actionを指定することで挙動を変えながらまとめてリクエストを投げることができます。

Provides a way to perform multiple index, create, delete, and update actions in a single request. The actions are specified in the request body using a newline delimited JSON (NDJSON) structure:

その中でも今回はindexアクションでの更新を選択しました。

index (Optional, string) Indexes the specified document. If the document exists, replaces the document and increments the version. The following line must contain the source data to be indexed.

ドキュメントには上記の記載がありました。

ここで一つ注意したいのが、すでにドキュメントがあったときに、indexを使用するとドキュメントの全てを更新してしまう点です。例えばid, name, priceというフィールドが存在したときにpriceだけ書き換えるというようなことはindexではできないのでご注意ください。

今回はindexでも対応できる要件でしたのでそのまま実装することにしました。

コードは以下です。

type Product struct {
    ID                int64                  `json:"id"`
    Name          string                 `json:"name"`
    Price            int64                  `json:"price"`
}

func BulkUpsert(products []*model.Product) error {
    es, err := elasticsearch.NewClient()
    if err != nil {
        log.Fatalf("Error creating the client: %s", err)
    }
    bi, err := esutil.NewBulkIndexer(esutil.BulkIndexerConfig{
        Index:  "items",
        Client: es,
    })
    if err != nil {
        log.Fatalf("Error creating the indexer: %s", err)
    }
    for _, a := range products {

        data, err := json.Marshal(a)
        if err != nil {
            log.Fatalf("json.Marshal(a) err %d: %s", a.ID, err)
        }
        bi.Add(
            context.Background(),
            esutil.BulkIndexerItem{
                // indexを指定
                Action:     "index",
                DocumentID: strconv.Itoa(int(a.ID)),
                Body:       bytes.NewReader(data),
            },
        )
    }
    return nil
}