Elasticsearch Ingest Node PipeLine 사용해 보기

---

Ingest Node를 이용하여 인덱싱하기 전에 전처리 설정 방법 입니다.

 

 

Elasticsearch 로컬 환경

Docker + Elasticsearch, Kibana 구성 (cluster)

 

현재 클러스터 노드 구성

# http://localhost:9200/_cat/nodes
172.25.0.2 36 3 0.05 0.49 0.44 dm * es1
172.25.0.3 36 3 0.05 0.49 0.44 dm - es2
172.25.0.4 36 3 0.05 0.49 0.44 d - es3

 

ingest node 추가

 

- ingest node 역할

          데이터 색인 시 전처리 작업인 pipeline 작업의 수행을 할 수 있는지 여부를 지정합니다.

          클러스터에 없을 경우 ingest pipeline 작업의 실행이 불가능합니다.

 

ingest node 없이 pipeline 실행

 

{
  "error" : {
    "root_cause" : [
      {
        "type" : "illegal_state_exception",
        "reason" : "There are no ingest nodes in this cluster, unable to forward request to an ingest node."
      }
    ],
    "type" : "illegal_state_exception",
    "reason" : "There are no ingest nodes in this cluster, unable to forward request to an ingest node."
  },
  "status" : 500
}

 

docker-compose.yml 파일에 es1 ingest node 추가

 

- node.roles=data,master,ingest
	
172.25.0.2 36 3 0.05 0.49 0.44 dmi * es1
172.25.0.3 36 3 0.05 0.49 0.44 dm - es2
172.25.0.4 36 3 0.05 0.49 0.44 d - es3

 

- pipeline 등록 

 

제공해주는 processor 사용 (split, html_strip)

 

          Kibana 메뉴 > Stack Management > Ingest > Ingest Node Pipelines

 

{
  "field": "splitList",
  "separator": ","
}

 

          명령어 등록

 

PUT _ingest/pipeline/ex2-pipeline
{
  "description": "html-strip ex",
  "processors": [
    {
      "html_strip": {
        "field": "contents"
      }
    }
  ]
}

 

# pipeline 조회
GET _ingest/pipeline


# 결과
{
  "ex2-pipeline" : {
    "description" : "html-strip ex",
    "processors" : [
      {
        "html_strip" : {
          "field" : "contents"
        }
      }
    ]
  },
  "ex1-pipeline" : {
    "description" : "split ex",
    "processors" : [
      {
        "split" : {
          "field" : "productList",
          "separator" : ","
        }
      }
    ]
  }
}

 

- pipeline 실행

 

# ex1-pipeline 사용
PUT pipeline_index/_doc/1?pipeline=ex1-pipeline
{
  "splitList": "test,test1,test2,test3,test4,test5"
}

# ex2-pipeline 사용
PUT pipeline_index/_doc/2?pipeline=ex2-pipeline
{
  "contents": "<h3>Hi</h3>"
}


# 결과
{
  "took" : 1,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 2,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "pipeline_index",
        "_type" : "_doc",
        "_id" : "1",
        "_score" : 1.0,
        "_source" : {
          "splitList" : [
            "test",
            "test1",
            "test2",
            "test3",
            "test4",
            "test5"
          ]
        }
      },
      {
        "_index" : "pipeline_index",
        "_type" : "_doc",
        "_id" : "2",
        "_score" : 1.0,
        "_source" : {
          "contents" : """
Hi
"""
        }
      }
    ]
  }
}

 

 

Script processor 사용

 

PUT _ingest/pipeline/ex3-pipeline
{
  "description": "script ex",
  "processors": [
    {
      "script": {
        "lang": "painless",
        "source": """
            ctx.testCode = ctx.testCode.replace("123", "");
        """
      }
    }
  ]
}

 

ex3-pipeline 사용해보기

 

{
  "took" : 410,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 4,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "pipeline_index",
        "_type" : "_doc",
        "_id" : "3",
        "_score" : 1.0,
        "_source" : {
          "testCode" : "Hi"
        }
      }
    ]
  }
}