[Golang/Elasticsearch] How to send data or anything to Elasticsearch by Golang

Posted on By nim No Comments on [Golang/Elasticsearch] How to send data or anything to Elasticsearch by Golang

Chúng ta cùng nhau sử dụng golang to send data vào elasticsearch.

Install Go Elasticsearch Library

You can install the Go Elasticsearch library using go get:

go get github.com/elastic/go-elasticsearch/v8

Create Go Code

Here is a Go code snippet that demonstrates how to create an index in Elasticsearch:

package main

import (
	"bytes"
	"context"
	"encoding/json"
	"log"

	"github.com/elastic/go-elasticsearch/v8"
	"github.com/elastic/go-elasticsearch/v8/esapi"
)

func main() {
	// Initialize Elasticsearch client
	cfg := elasticsearch.Config{
		Addresses: []string{
			"http://localhost:9200",
		},
	}
	es, err := elasticsearch.NewClient(cfg)
	if err != nil {
		log.Fatalf("Error initializing Elasticsearch client: %s", err)
	}

	// Define the index mapping
	indexMapping := map[string]interface{}{
		"mappings": map[string]interface{}{
			"properties": map[string]interface{}{
				"name": map[string]interface{}{
					"type": "text",
				},
				"age": map[string]interface{}{
					"type": "integer",
				},
			},
		},
	}

	// Convert map to JSON
	mappingJSON, err := json.Marshal(indexMapping)
	if err != nil {
		log.Fatalf("Error marshaling index mapping: %s", err)
	}

	// Create the index
	req := esapi.IndicesCreateRequest{
		Index: "my_index",
		Body:  bytes.NewReader(mappingJSON),
	}

	res, err := req.Do(context.Background(), es)
	if err != nil {
		log.Fatalf("Error creating index: %s", err)
	}
	defer res.Body.Close()

	if res.IsError() {
		log.Fatalf("Error: %s", res.String())
	}

	log.Println("Index created successfully")
}

Nếu bạn muốn xem mappingJSON là cái j?

fmt.Println(string(mappingJSON))
{
  "mappings": {
    "properties": {
      "age": {
        "type": "integer"
      },
      "name": {
        "type": "text"
      }
    }
  }
}

Apply Singleton with Elasticsearch and Golang.

1) Create Index in Elasticsearch by Golang

Đầu tiên mình sẽ define riêng 1 folder helper/search

Ở helper/search/elasticEngine.go
Mình sẽ thực hiện tạo 1 connection và retry nếu có lỗi:

package search

import (
	"beerus-land/log"
	"github.com/elastic/go-elasticsearch/v8"
	"time"
)

type ElasticEngine struct {
	EE            *elasticsearch.Client
	ConnectString string
}

func (ee *ElasticEngine) Connect() error {
	cfg := elasticsearch.Config{
		Addresses: []string{
			ee.ConnectString,
		},
	}

	maxRetries := 3               // Maximum number of retry attempts
	retryDelay := 5 * time.Second // Delay between retry attempts

	var err error
	for i := 0; i < maxRetries; i++ {
		ee.EE, err = elasticsearch.NewClient(cfg)
		if err == nil {
			log.Info("Connected to ElasticSearch")
			return nil // Connection successful
		}

		log.Error("Failed to connect to ElasticSearch: ", err)
		log.Info("Retrying in ", retryDelay)
		time.Sleep(retryDelay) // Wait before retrying
	}

	log.Error("Max retries reached. Failing...")
	return err // Return the last error
}

func (ee *ElasticEngine) Close() {
	if ee.EE != nil && ee.EE.Transport != nil {
		if transport, ok := ee.EE.Transport.(interface{ CloseIdleConnections() }); ok {
			transport.CloseIdleConnections()
		}
	}
	ee.EE = nil
}

Tiếp đến là helper/search/search_impl
Đây là nơi mà mình design các function và public chúng cho handler:

helper/search/search_impl/elasticEngine_impl.go

package search_impl

import "beerus-land/helper/search"

type ElasticEngineImpl struct {
	elasticEngine *search.ElasticEngine
}

func NewLanderElastic(elasticEngine *search.ElasticEngine) search.LanderElastic {
	return &ElasticEngineImpl{
		elasticEngine: elasticEngine,
	}
}

helper/search/search_impl/landAndHouse_search_impl.go
Chúng ta sẽ cùng đi vào function đầu tiên là create Index trên elastichsearch

package search_impl

import (
	"beerus-land/log"
	"beerus-land/model"
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"github.com/elastic/go-elasticsearch/v8/esapi"
	"time"
)

func (eei *ElasticEngineImpl) CreateSearchIndices(name string) (model.Indices, error) {

	indeX := model.Indices{}

	formattedIndexName := name + "-" + time.Now().Format("2006.01.02")
	// Create a context for the API calls
	ctx := context.Background()

	// Check if the index already exists
	existsReq := esapi.IndicesExistsRequest{
		Index: []string{formattedIndexName},
	}
	existsRes, err := existsReq.Do(ctx, eei.elasticEngine.EE)
	if err != nil {
		log.Error(err.Error())
		return indeX, err
	}
	defer existsRes.Body.Close()

	if existsRes.StatusCode == 200 {
		indeX = model.Indices{IndexName: formattedIndexName}
		log.Info("Index " + formattedIndexName + " already exists")
		return indeX, nil
	}

	// Define the mapping for HouseAndLand
	mapping := map[string]interface{}{
		"mappings": map[string]interface{}{
			"properties": map[string]interface{}{
				"time-stamp": map[string]interface{}{
					"type": "date",
				},
				"city": map[string]interface{}{
					"type": "text",
					"fields": map[string]interface{}{
						"keyword": map[string]interface{}{
							"type": "keyword",
						},
					},
				},
				"district": map[string]interface{}{
					"type": "text",
					"fields": map[string]interface{}{
						"keyword": map[string]interface{}{
							"type": "keyword",
						},
					},
				},
				"ward": map[string]interface{}{
					"type": "text",
					"fields": map[string]interface{}{
						"keyword": map[string]interface{}{
							"type": "keyword",
						},
					},
				},
				"post-title": map[string]interface{}{
					"type": "text",
					"fields": map[string]interface{}{
						"keyword": map[string]interface{}{
							"type": "keyword",
						},
					},
				},
				"land-area": map[string]interface{}{
					"type": "float",
				},
				"price": map[string]interface{}{
					"type": "float",
				},
				"post_owner": map[string]interface{}{
					"type": "text",
					"fields": map[string]interface{}{
						"keyword": map[string]interface{}{
							"type": "keyword",
						},
					},
				},
				"link": map[string]interface{}{
					"type": "text",
					"fields": map[string]interface{}{
						"keyword": map[string]interface{}{
							"type": "keyword",
						},
					},
				},
				"priority": map[string]interface{}{
					"type": "keyword",
				},
				"number_of_bedroom": map[string]interface{}{
					"type": "integer",
				},
			},
		},
	}

	// Convert the mapping to JSON
	var buf bytes.Buffer
	if err := json.NewEncoder(&buf).Encode(mapping); err != nil {
		log.Fatalf("Error encoding mapping: %s", err)
	}

	req := esapi.IndicesCreateRequest{
		Index: formattedIndexName,
		Body:  &buf,
	}

	// Execute the index creation request
	res, err := req.Do(ctx, eei.elasticEngine.EE)
	if err != nil {
		log.Error(err.Error())
		return indeX, err
	}
	defer res.Body.Close()

	// Check the response status
	if res.IsError() {
		log.Error("Error creating index: %s", res.Status())
		return indeX, fmt.Errorf("Error creating index: %s", res.Status())
	}

	indeX = model.Indices{IndexName: formattedIndexName}
	log.Info("Index " + formattedIndexName + " created successfully")
	return indeX, nil
}

Ở phần này bạn sẽ cần hiểu nhất về mapping trong khi tạo index
Bạn đọc bài bên dưới

[Golang] Define mapping when creating an index on Elasticsearch by Golang.

Sau khi đã code xong function create index thì mình cần public nó ra:
helper/search/landerElastic.go

package search

import (
	"beerus-land/model"
	"time"
)

type LanderElastic interface {
	CreateSearchIndices(name string) (model.Indices, error)
}

ở file main.yaml

package main

import (
	"beerus-land/handler"
	"beerus-land/helper/automation/automation_impl"
	"beerus-land/helper/search"
	"beerus-land/helper/search/search_impl"
	"beerus-land/log"
	"github.com/go-co-op/gocron"
	"github.com/labstack/echo/v4"
	"os"
	"time"
)

func init() {
	os.Setenv("APP_NAME", "BeerUs LanD")
	log.InitLogger(false)
	os.Setenv("TZ", "Asia/Ho_Chi_Minh")
}

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

	elas := &search.ElasticEngine{
		ConnectString: "http://192.168.101.25:9200",
	}
	elas.Connect()
	defer elas.Close()

	autoCrawHandler := handler.AutoCrawHandler{
		Echo:              nil,
		ElasticEngineImpl: search_impl.NewLanderElastic(elas),
	}

	s := gocron.NewScheduler(time.UTC)
	s.Cron("*/5 * * * *").Do(func() { autoCrawHandler.HandlerCrawPrice() })
	s.StartAsync()
	autoCrawHandler.HandlerCrawPrice()

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

Và cuối cùng là trong handler;
handler/autoCraw_handler.go

package handler

import (
	"beerus-land/helper/automation"
	"beerus-land/helper/notify"
	"beerus-land/helper/search"
	"beerus-land/log"
	"beerus-land/model"
	"fmt"
	"github.com/labstack/echo/v4"
	"time"
)

type AutoCrawHandler struct {
	Echo              *echo.Context
	ElasticEngineImpl search.LanderElastic
}

func (a *AutoCrawHandler) HandlerCrawPrice() error {
	lookUp := model.HouseAndLand{
		City:     "tp-ho-chi-minh",
		District: "quan-go-vap",
	}
	crawData, err := a.Automation.CrawPrice(lookUp)
	if err != nil {
		log.Error(err.Error())
		return err
	}
	//fmt.Printf("%+v\n", crawData)

	index, err := a.ElasticEngineImpl.CreateSearchIndices("land-house")

...
...

2) Push data to Elasticsearch.

ELK

More Related Articles

[error] Solve elasticsearch error Limit of total fields [1000] in index [xxx] ELK
[logstash/kibana] Draw a Map on Kibana by data from Logstash – Fix “Index pattern does not contain any geospatial fields” ELK
[Fortigate] Cấu hình firewall fortigate gửi syslog sang Logstash BareMetal
[Elastic] Install Elasticsearch through package manager. ELK
[Elasticsearch] reason=Validation Failed: 1: this action would add [10] total shards, but this cluster currently has [1000]/[1000] maximum shards open;]]] ELK
[Golang] Define mapping when creating an index on Elasticsearch by Golang. ELK

Leave a Reply

Your email address will not be published. Required fields are marked *