OpenSearch에서 JSON Processor를 사용하여 JSON문자열을 JSON으로 변환하는 방법

Ingest Processors란?

OpenSearch는 데이터 색인 및 검색을 위한 강력한 오픈소스 플랫폼으로, 다양한 기능을 제공합니다. 그 중에서도 Ingest Processors는 데이터 전처리 과정에서 중요한 역할을 합니다.

Ingest Processors는 OpenSearch에서 데이터를 색인하기 전 문서를 전처리하는 도구입니다. 이를 통해 필드 제거, 텍스트에서 값 추출, 데이터 형식 변환, 추가 정보 첨부 등 다양한 작업을 수행할 수 있습니다. [1]

OpenSearch에서 지원 되는 Processors

Opensearch에서는 총 40개의 프로세서를 지원합니다. 그중 대표적으로 아래의 processors를 소개해드립니다.

• date: 문서에서 날짜나 타임스탬프 필드를 파싱하고, 이를 문서의 타임스탬프로 사용합니다.
• grok: 패턴 매칭을 사용하여 비정형 데이터를 파싱하고 구조화합니다.
• geoip: IP 주소에 대한 지리적 위치 정보를 추가합니다.
• lowercase/uppercase: 특정 필드의 텍스트를 소/대문자로 변환합니다.
• remove: 문서에서 필드를 제거합니다.

Ingest Processors 사용 방법

Ingest Processors 사용하기 위해서는 OpenSearch에서 Ingest Processors를 사용하려면 다음과 같은 단계를 따릅니다:

1. 사용 가능한 프로세서 확인: Nodes Info API를 사용하여 현재 설치된 OpenSearch에서 사용 가능한 프로세서 목록을 확인합니다.

 GET /_nodes/ingest?filter_path=nodes.*.ingest.processors    

2. 파이프라인 생성: 원하는 프로세서를 포함한 Ingest 파이프라인을 생성합니다.
3. 데이터 색인 시 파이프라인 적용: 문서를 색인할 때 생성한 파이프라인을 지정하여 전처리를 수행합니다.

JSON Processor 사용

Opensearch는 다양한 processor를 제공하지만, 그 중 JSON processor가 유용합니다. 이 processor는 문자열 형태의 JSON을 구조화된 JSON 객체로 변환하는 데 사용됩니다. JSON 형식의 데이터를 포함한 필드를 처리할 때 특히 효과적으로, 데이터의 구조화와 접근성을 크게 향상시킵니다.

파이프라인 생성 및 기존 문서에 적용하는 방법

1. 파이프라인 생성: 이 단계는 데이터 처리를 위한 파이프라인을 정의합니다.

   PUT _ingest/pipeline/<pipeline_name>
   {
     "description": "<pipeline description>",
     "processors": [
       {
         "<processor_type>": {
           "field": "<source_field>",
           "target_field": "<target_field>"
         }
       }
     ]
   }
   

   • <pipeline_name>: 생성할 파이프라인의 이름
   • <Pipeline Description>: 파이프라인의 용도나 기능을 설명하는 문자열
   • <processor_type>: 사용할 프로세서의 유형 (예: json, split, date 등)
   • <source_field>: 처리할 원본 필드 이름
   • <target_field>: 처리 결과를 저장할 대상 필드 이름

2. 파이프라인 적용: 이 단계는 생성한 파이프라인을 기존 문서에 적용합니다.

   POST <index_name>/_update_by_query?pipeline=<pipeline_name>
   {
     "query": {
       "exists": {
         "field": "<field_name>"
       }
     }
   }
   

   • <index_name>: 파이프라인을 적용할 인덱스 이름
   • <pipeline_name>: 적용할 파이프라인의 이름
   • <field_name>: 파이프라인을 적용할 조건이 되는 필드 이름

예시: JSON Processor를 이용하여 people 인덱스 전처리

우선 people 인덱스를 생성하며, name, age, languages 필드를 포함합니다. languages 필드에는 JSON 문자열 형태의 데이터가 저장됩니다. JSON Processor를 사용하여 languages 필드의 JSON 문자열을 구조화된 JSON 객체로 변환합니다.

1. 인덱스 생성 및 데이터 삽입

• 'people'이라는 인덱스를 생성하고 샘플 데이터를 삽입합니다.

PUT /people
{
  "mappings": {
    "properties": {
      "name": { "type": "keyword" },
      "age": { "type": "integer" },
      "languages": { "type": "text" }
    }
  }
}

POST /people/_doc
{
  "name": "Davis",
  "age": 30,
  "languages": "{\"language1\":\"English\",\"language2\":\"Korean\"}"
}

2. 파이프라인 생성 : 여기서 'languages' 필드는 JSON 형식의 문자열로 저장됩니다.

• JSON 파싱 파이프라인 생성 다음으로, JSON 문자열을 파싱하는 파이프라인을 생성합니다.
• 'languages' 필드의 JSON 문자열을 파싱하여 'parsed_languages'라는 새 필드로 저장됩니다.

PUT _ingest/pipeline/json_pipeline
{
  "description": "Pipeline for parsing JSON strings",
  "processors": [
    {
      "json": {
        "field": "languages",
        "target_field": "parsed_languages"
      }
    }
  ]
}

3. 파이프라인을 기존 문서에 적용합니다.

• 이 명령은 people인덱스의 'languages' 필드가 존재하는 모든 문서에 파이프라인을 업데이트 합니다.

POST people/_update_by_query?pipeline=json_pipeline
{
  "query": {
    "exists": {
      "field": "languages"
    }
  }
}

4. 결과 확인 : 변환된 데이터를 확인합니다.

GET people/_search

참고

[1] Ingest processors https://opensearch.org/docs/latest/ingest-pipelines/processors/index-processors/