Go 言語+Elasticsearch で簡易求人サイトバックエンド構築

.
├── README.md
├── batch
│   ├── LoadData.go
│   ├── go.mod
│   ├── go.sum
│   └── test_data.xml
├── docker-compose.yml
├── es
│   ├── dic
│   │   └── test_dic.csv
│   ├── Dockerfile
│   ├── script
│   │   └── es_init.sh
│   └── sudachi
│       └── sudachi.json
├── search_api
│   ├── Dockerfile
│   ├── go.mod
│   ├── go.sum
│   ├── hr_api
│   ├── internal
│   │   ├── connect_es.go
│   │   ├── hr_query.go
│   │   └── hr_search.go
│   └── main.go

 #東京都の"カフェ"の求人を検索する
http://localhost:5000/search?keyword=カフェ&state=東京都

 #東京都の"Go言語"の求人を検索する
http://localhost:5000/search?keyword=Go言語&state=東京都

 #神奈川県の"アルバイト・パート"の求人を検索する
http://localhost:5000/search?keyword=アルバイト・パート&state=神奈川県

 #求人のユニークidから検索する
http://localhost:5000/search?id=test

ARG ELASTIC_VER=7.8.1
ARG SUDACHI_PLUGIN_VER=2.0.3

FROM ibmjava:8-jre-alpine as dict_builder

ARG ELASTIC_VER
ARG SUDACHI_PLUGIN_VER

WORKDIR /home

RUN wget https://github.com/WorksApplications/Elasticsearch-sudachi/releases/download/v${ELASTIC_VER}-${SUDACHI_PLUGIN_VER}/analysis-sudachi-${ELASTIC_VER}-${SUDACHI_PLUGIN_VER}.zip && \
    unzip analysis-sudachi-${ELASTIC_VER}-${SUDACHI_PLUGIN_VER}.zip && \
    wget http://sudachi.s3-website-ap-northeast-1.amazonaws.com/sudachidict/sudachi-dictionary-20210802-core.zip && \
    unzip sudachi-dictionary-20210802-core.zip && \
    mkdir -p /usr/share/Elasticsearch/config/sudachi/ && \
    mv sudachi-dictionary-20210802/system_core.dic /usr/share/Elasticsearch/config/sudachi/ && \
    rm -rf sudachi-dictionary-20210802-core.zip sudachi-dictionary-20210802/


FROM docker.elastic.co/Elasticsearch/Elasticsearch:${ELASTIC_VER}

ARG ELASTIC_VER
ARG SUDACHI_PLUGIN_VER

COPY es/sudachi/sudachi.json /usr/share/Elasticsearch/config/sudachi/
COPY --from=dict_builder /home/analysis-sudachi-${ELASTIC_VER}-${SUDACHI_PLUGIN_VER}.zip /usr/share/Elasticsearch/

  Elasticsearch:
    build:
      context: .
      dockerfile: es/dockerfile
    container_name: Elasticsearch
    volumes:
      - es-data:/usr/share/Elasticsearch/data
    networks:
      - Elasticsearch
    ports:
      - 9200:9200
    environment:
      - discovery.type=single-node
      - node.name=Elasticsearch
      - cluster.name=go-Elasticsearch-docker-cluster
      - bootstrap.memory_lock=true
      - ES_JAVA_OPTS=-Xms256m -Xmx256m
    ulimits:
      { nofile: { soft: 65535, hard: 65535 }, memlock: { soft: -1, hard: -1 } }
    healthcheck:
      test: curl --head --max-time 120 --retry 120 --retry-delay 1 --show-error --silent http://localhost:9200

  kibana:
    container_name: kibana
    image: docker.elastic.co/kibana/kibana:7.8.1
    depends_on: ["Elasticsearch"]
    networks:
      - Elasticsearch
    ports:
      - 5601:5601
    environment:
      - Elasticsearch_HOSTS=http://Elasticsearch:9200
      - KIBANA_LOGGING_QUIET=true
    healthcheck:
      test: curl --max-time 120 --retry 120 --retry-delay 1 --show-error --silent http://localhost:5601

req := esapi.IndexRequest{
                Index:      "baito",
                DocumentID: string(j.Referencenumber),
                Body:       strings.NewReader(string(jobbody)),
                Refresh:    "true",
            }

package main

import (
    "bytes"
    "encoding/json"
    "encoding/xml"
    "flag"
    "fmt"
    "io"
    "log"
    "math/rand"
    "os"
    "strings"
    "time"

    "github.com/dustin/go-humanize"
    "github.com/elastic/go-Elasticsearch/v7"
    "github.com/elastic/go-Elasticsearch/v7/esapi"
    "github.com/joho/godotenv"
)

type Job struct {
    Referencenumber string `xml:"referencenumber" json:"referencenumber,string"`
    Date            string `xml:"date" json:"date,string"`
    Url             string `xml:"url" json:"url,string"`
    Title           string `xml:"title" json:"title,string"`
    Description     string `xml:"description" json:"description,string"`
    State           string `xml:"state" json:"state,string"`
    City            string `xml:"city" json:"city,string"`
    Country         string `xml:"country" json:"country,string"`
    Station         string `xml:"station" json:"station,string"`
    Jobtype         string `xml:"jobtype" json:"jobtype,string"`
    Salary          string `xml:"salary" json:"salary,string"`
    Category        string `xml:"category" json:"category,string"`
    ImageUrls       string `xml:"imageUrls" json:"imageurls,string"`
    Timeshift       string `xml:"timeshift" json:"timeshift,string"`
    Subwayaccess    string `xml:"subwayaccess" json:"subwayaccess,string"`
    Keywords        string `xml:"keywords" json:"keywords,string"`
}

var (
    _     = fmt.Print
    count int
    batch int
)

func init() {
    flag.IntVar(&count, "count", 300000, "Number of documents to generate")
    flag.IntVar(&batch, "batch", 1000, "Number of documents to send in one batch")
    flag.Parse()

    rand.Seed(time.Now().UnixNano())
}

func main() {


    log.SetFlags(0)

        type bulkResponse struct {
        Errors bool `json:"errors"`
        Items  []struct {
            Index struct {
                ID     string `json:"_id"`
                Result string `json:"result"`
                Status int    `json:"status"`
                Error  struct {
                    Type   string `json:"type"`
                    Reason string `json:"reason"`
                    Cause  struct {
                        Type   string `json:"type"`
                        Reason string `json:"reason"`
                    } `json:"caused_by"`
                } `json:"error"`
            } `json:"index"`
        } `json:"items"`
    }

        var (
        buf bytes.Buffer
        res *esapi.Response
        err error
        raw map[string]interface{}
        blk *bulkResponse

        jobs  []*Job
        indexName = "baito"

        numItems   int
        numErrors  int
        numIndexed int
        numBatches int
        currBatch  int
    )

    log.Printf(
    "\x1b[1mBulk\x1b[0m: documents [%s] batch size [%s]",
    humanize.Comma(int64(count)), humanize.Comma(int64(batch)))
    log.Println(strings.Repeat("▁", 65))

    // Create the Elasticsearch client
    //
    es, err := Elasticsearch.NewDefaultClient()
    if err != nil {
        log.Fatalf("Error creating the client: %s", err)
    }

    err = godotenv.Load(".env")
    if err != nil {
        log.Fatal("Error loading .env file")
    }
    xml_path := os.Getenv("BAITO_XML_PATH")
    f, err := os.Open(xml_path)

    if err != nil {
        log.Fatal(err)
    }

    defer f.Close()

    d := xml.NewDecoder(f)

    for i := 1; i < count+1; i++ {
        t, tokenErr := d.Token()
        if tokenErr != nil {
            if tokenErr == io.EOF {
                break
            }
            // handle error somehow
            log.Fatalf("Error decoding token: %s", tokenErr)
        }
        switch ty := t.(type) {
        case xml.StartElement:
            if ty.Name.Local == "job" {
                // If this is a start element named "location", parse this element
                // fully.
                var job Job
                if err = d.DecodeElement(&job, &ty); err != nil {
                    log.Fatalf("Error decoding item: %s", err)
                } else {
                    jobs = append(jobs, &job)
                }
            }
        default:
        }
        // fmt.Println("count =", count)
    }
    log.Printf("→ Generated %s articles", humanize.Comma(int64(len(jobs))))
    fmt.Print("→ Sending batch ")

        // Re-create the index
    //
    if res, err = es.Indices.Delete([]string{indexName}); err != nil {
        log.Fatalf("Cannot delete index: %s", err)
    }
    res, err = es.Indices.Create(indexName)
    if err != nil {
        log.Fatalf("Cannot create index: %s", err)
    }
    if res.IsError() {
        log.Fatalf("Cannot create index: %s", res)
    }

    if count%batch == 0 {
        numBatches = (count / batch)
    } else {
        numBatches = (count / batch) + 1
    }

    start := time.Now().UTC()

    // Loop over the collection
    //
    for i, a := range jobs {
        numItems++

        currBatch = i / batch
        if i == count-1 {
            currBatch++
        }

        // Prepare the metadata payload
        //
        meta := []byte(fmt.Sprintf(`{ "index" : { "_id" : "%d" } }%s`, a.Referencenumber, "\n"))
        // fmt.Printf("%s", meta) // <-- Uncomment to see the payload

        // Prepare the data payload: encode article to JSON
        //
        data, err := json.Marshal(a)
        if err != nil {
            log.Fatalf("Cannot encode article %d: %s", a.Referencenumber, err)
        }

        // Append newline to the data payload
        //
        data = append(data, "\n"...) // <-- Comment out to trigger failure for batch
        // fmt.Printf("%s", data) // <-- Uncomment to see the payload

        // // Uncomment next block to trigger indexing errors -->
        // if a.ID == 11 || a.ID == 101 {
        //  data = []byte(`{"published" : "INCORRECT"}` + "\n")
        // }
        // // <--------------------------------------------------

        // Append payloads to the buffer (ignoring write errors)
        //
        buf.Grow(len(meta) + len(data))
        buf.Write(meta)
        buf.Write(data)

        // When a threshold is reached, execute the Bulk() request with body from buffer
        //
        if i > 0 && i%batch == 0 || i == count-1 {
            fmt.Printf("[%d/%d] ", currBatch, numBatches)

            res, err = es.Bulk(bytes.NewReader(buf.Bytes()), es.Bulk.WithIndex(indexName))
            if err != nil {
                log.Fatalf("Failure indexing batch %d: %s", currBatch, err)
            }
            // If the whole request failed, print error and mark all documents as failed
            //
            if res.IsError() {
                numErrors += numItems
                if err := json.NewDecoder(res.Body).Decode(&raw); err != nil {
                    log.Fatalf("Failure to to parse response body: %s", err)
                } else {
                    log.Printf("  Error: [%d] %s: %s",
                        res.StatusCode,
                        raw["error"].(map[string]interface{})["type"],
                        raw["error"].(map[string]interface{})["reason"],
                    )
                }
                // A successful response might still contain errors for particular documents...
                //
            } else {
                if err := json.NewDecoder(res.Body).Decode(&blk); err != nil {
                    log.Fatalf("Failure to to parse response body: %s", err)
                } else {
                    for _, d := range blk.Items {
                        // ... so for any HTTP status above 201 ...
                        //
                        if d.Index.Status > 201 {
                            // ... increment the error counter ...
                            //
                            numErrors++

                            // ... and print the response status and error information ...
                            log.Printf("  Error: [%d]: %s: %s: %s: %s",
                                d.Index.Status,
                                d.Index.Error.Type,
                                d.Index.Error.Reason,
                                d.Index.Error.Cause.Type,
                                d.Index.Error.Cause.Reason,
                            )
                        } else {
                            // ... otherwise increase the success counter.
                            //
                            numIndexed++
                        }
                    }
                }
            }

            // Close the response body, to prevent reaching the limit for goroutines or file handles
            //
            res.Body.Close()

            // Reset the buffer and items counter
            //
            buf.Reset()
            numItems = 0
        }
    }

    // Report the results: number of indexed docs, number of errors, duration, indexing rate
    //
    fmt.Print("\n")
    log.Println(strings.Repeat("▔", 65))

    dur := time.Since(start)

    if numErrors > 0 {
        log.Fatalf(
            "Indexed [%s] documents with [%s] errors in %s (%s docs/sec)",
            humanize.Comma(int64(numIndexed)),
            humanize.Comma(int64(numErrors)),
            dur.Truncate(time.Millisecond),
            humanize.Comma(int64(1000.0/float64(dur/time.Millisecond)*float64(numIndexed))),
        )
    } else {
        log.Printf(
            "Sucessfuly indexed [%s] documents in %s (%s docs/sec)",
            humanize.Comma(int64(numIndexed)),
            dur.Truncate(time.Millisecond),
            humanize.Comma(int64(1000.0/float64(dur/time.Millisecond)*float64(numIndexed))),
        )
    }
}

for job in jobs:

    index = job.as_dict()
    if job.description == "" or job.description == null:
        continue

    bulk_file += json.dumps(
        {"index": {"_index": index_name, "_type": "_doc", "_id": id}}
    )

    # The optional_document portion of the bulk file
    bulk_file += "\n" + json.dumps(index) + "\n"

    if id % 1000 == 0:
        response = client.bulk(bulk_file)
        bulk_file = ""
        id += 1
        continue

    id += 1
if bulk_file != "":
    response = client.bulk(bulk_file)

.
├── search_api
│   ├── Dockerfile
│   ├── go.mod
│   ├── go.sum
│   ├── hr_api
│   ├── internal
│   │   ├── connect_es.go
│   │   ├── hr_query.go
│   │   └── hr_search.go
│   └── main.go

package main

import (
    internal "hr_api/internal"

    "github.com/labstack/echo"
    "github.com/labstack/echo/middleware"
)

func main() {
    e := echo.New()

    e.Use(middleware.Logger())
    e.Use(middleware.Recover())
    e.Use(middleware.CORS())

    e.GET("/search", internal.HRSearch)

    e.Logger.Fatal(e.Start(":5000"))
}

package internal

import (
    "bytes"
    "context"
    "encoding/json"
    "fmt"
    "net/http"

    "github.com/labstack/echo"
)

type Query struct {
    Keyword string `query:"keyword"`
    State   string `query:"state"`
    Id      string `query:"id"`
}

type Result struct {
    Referencenumber string `xml:"referencenumber" json:"referencenumber,string"`
    Date            string `xml:"date" json:"date,string"`
    Url             string `xml:"url" json:"url,string"`
    Title           string `xml:"title" json:"title,string"`
    Description     string `xml:"description" json:"description,string"`
    State           string `xml:"state" json:"state,string"`
    City            string `xml:"city" json:"city,string"`
    Country         string `xml:"country" json:"country,string"`
    Station         string `xml:"station" json:"station,string"`
    Jobtype         string `xml:"jobtype" json:"jobtype,string"`
    Salary          string `xml:"salary" json:"salary,string"`
    Category        string `xml:"category" json:"category,string"`
    ImageUrls       string `xml:"imageUrls" json:"imageurls,string"`
    Timeshift       string `xml:"timeshift" json:"timeshift,string"`
    Subwayaccess    string `xml:"subwayaccess" json:"subwayaccess,string"`
    Keywords        string `xml:"keywords" json:"keywords,string"`
}
type Response struct {
    Message string `json:"message"`
    Results []Result
}

func HRSearch(c echo.Context) (err error) {
    // クライアントからのパラメーターを取得
    q := new(Query)
    if err = c.Bind(q); err != nil {
        return
    }

    res := new(Response)
    var (
        b   map[string]interface{}
        buf bytes.Buffer
    )

    // Elasticsearch へのクエリを作成
    query := CreateQuery(q)

    json.NewEncoder(&buf).Encode(query)

    fmt.Printf(buf.String())

    // Elasticsearch へ接続
    es, err := ConnectElasticsearch()
    if err != nil {
        c.Error(err)
    }

    // Elasticsearch へクエリ
    r, err := es.Search(
        es.Search.WithContext(context.Background()),
        es.Search.WithIndex("baito"),
        es.Search.WithBody(&buf),
        es.Search.WithTrackTotalHits(true),
        es.Search.WithPretty(),
    )
    if err != nil {
        c.Error(err)
    }
    defer r.Body.Close()

    if err := json.NewDecoder(r.Body).Decode(&b); err != nil {
        c.Error(err)
    }

    // クエリの結果を Responce.Results に格納
    for _, hit := range b["hits"].(map[string]interface{})["hits"].([]interface{}) {
        result := new(Result)
        doc := hit.(map[string]interface{})

        fmt.Printf(result.Title)

        result.Referencenumber = doc["_source"].(map[string]interface{})["referencenumber"].(string)
        result.Date = doc["_source"].(map[string]interface{})["date"].(string)
        result.Url = doc["_source"].(map[string]interface{})["url"].(string)
        result.Title = doc["_source"].(map[string]interface{})["title"].(string)
        result.State = doc["_source"].(map[string]interface{})["state"].(string)
        result.Category = doc["_source"].(map[string]interface{})["category"].(string)
        result.Description = doc["_source"].(map[string]interface{})["description"].(string)
        result.City = doc["_source"].(map[string]interface{})["city"].(string)
        result.Country = doc["_source"].(map[string]interface{})["country"].(string)
        result.Station = doc["_source"].(map[string]interface{})["station"].(string)
        result.Jobtype = doc["_source"].(map[string]interface{})["jobtype"].(string)
        result.Salary = doc["_source"].(map[string]interface{})["salary"].(string)
        result.ImageUrls = doc["_source"].(map[string]interface{})["imageurls"].(string)
        result.Timeshift = doc["_source"].(map[string]interface{})["timeshift"].(string)
        result.Subwayaccess = doc["_source"].(map[string]interface{})["subwayaccess"].(string)
        result.Keywords = doc["_source"].(map[string]interface{})["keywords"].(string)

        res.Results = append(res.Results, *result)
    }

    res.Message = "検索に成功しました。"

    return c.JSON(http.StatusOK, res)
}

package internal

func CreateQuery(q *Query) map[string]interface{} {
  query := map[string]interface{}{}
  if q.Id != "" {
    query = map[string]interface{}{
      "query": map[string]interface{}{
        "bool": map[string]interface{}{
          "must": []map[string]interface{}{
            {
              "match": map[string]interface{}{
                "referencenumber": q.Id,
              },
            },
          },
        },
      },
    }
  } else if q.Keyword != "" && q.State != "" {
    query = map[string]interface{}{
      "query": map[string]interface{}{
        "bool": map[string]interface{}{
          "must": []map[string]interface{}{
            {
              "bool": map[string]interface{}{
                "should": []map[string]interface{}{
                  {
                    "match": map[string]interface{}{
                      "title": map[string]interface{}{
                        "query": q.Keyword,
                        "boost": 3,
                      },
                    },
                  },
                  {
                    "match": map[string]interface{}{
                      "description": map[string]interface{}{
                        "query": q.Keyword,
                        "boost": 2,
                      },
                    },
                  },
                  {
                    "match": map[string]interface{}{
                      "category": map[string]interface{}{
                        "query": q.Keyword,
                        "boost": 1,
                      },
                    },
                  },
                },
                "minimum_should_match": 1,
              },
            },
            {
              "bool": map[string]interface{}{
                "must": []map[string]interface{}{
                  {
                    "match": map[string]interface{}{
                      "state": q.State,
                    },
                  },
                },
              },
            },
          },
        },
      },
    }
  } else if q.Keyword != "" && q.State == "" {
    query = map[string]interface{}{
      "query": map[string]interface{}{
        "bool": map[string]interface{}{
          "should": []map[string]interface{}{
            {
              "match": map[string]interface{}{
                "title": map[string]interface{}{
                  "query": q.Keyword,
                  "boost": 3,
                },
              },
            },
            {
              "match": map[string]interface{}{
                "description": map[string]interface{}{
                  "query": q.Keyword,
                  "boost": 2,
                },
              },
            },
            {
              "match": map[string]interface{}{
                "category": map[string]interface{}{
                  "query": q.Keyword,
                  "boost": 1,
                },
              },
            },
          },
          "minimum_should_match": 1,
        },
      },
    }
  } else if q.Keyword == "" && q.State != "" {
    query = map[string]interface{}{
      "query": map[string]interface{}{
        "bool": map[string]interface{}{
          "must": []map[string]interface{}{
            {
              "match": map[string]interface{}{
                "state": q.State,
              },
            },
          },
        },
      },
    }
  }

  return query
}

package internal

import (
    "os"

    Elasticsearch "github.com/elastic/go-Elasticsearch/v7"
)

func ConnectElasticsearch() (*Elasticsearch.Client, error) {
    // 環境変数 ES_ADDRESS がある場合は記述されているアドレスに接続
    // ない場合は、 http://localhost:9200 に接続
    var addr string
    if os.Getenv("ES_ADDRESS") != "" {
        addr = os.Getenv("ES_ADDRESS")
    } else {
        addr = "http://localhost:9200"
    }
    cfg := Elasticsearch.Config{
        Addresses: []string{
            addr,
        },
    }
    es, err := Elasticsearch.NewClient(cfg)

    return es, err
}

http://localhost:5000/search?keyword=カフェ&state=東京都

{
    "message": "検索に成功しました。",
    "Results": [
        {
            "referencenumber": "test",
            "date": "2222-11-01",
            "url": "test",
            "title": "おしゃれカフェ・店舗スタッフ/ブック＆カフェ/アルバイト・パート/おしゃれカフェ",
            "description": "【省略】",
            "state": "東京都",
            "city": "渋谷区",
            "country": "日本",
            "station": "山手線渋谷駅 徒歩700分",
            "jobtype": "アルバイト・パート",
            "salary": "test円",
            "category": "飲食・フード×おしゃれカフェ",
            "imageurls": "test",
            "timeshift": "週3日以上/1日3時間以上",
            "subwayaccess": "山手線渋谷駅徒歩700分",
            "keywords": "test"
        },