[Elasticsearch + Node.js] index 생성 삭제 조회 bulk CRUD with TypeScript

 

Elasticsearch + Node.js + Typescript로 elasticsearch 실습하기 

 

@elastic/elasticsearch version : 8.1.0

typescript version : 4.6.3

 

 

 

1. 새 프로젝트 환경 설정

https://blckchainetc.tistory.com/381

[TypeScript + Node.js] 새 프로젝트 시작하기 (환경 설정, nodemon, rimraf)

 

 

2. Elasticsearch와 연결하기 

https://blckchainetc.tistory.com/382

[Elasticsearch + Node.js] with Typescript 연결하기

 

 

3. index.ts 파일에 실습해보기 

 

1) index 에 document 생성하기 - 해당 인덱스가 없다면 자동으로 생성한다

import { Client } from '@elastic/elasticsearch';

const client = new Client({
    node: 'http://localhost:9200'
})

const run = async() => {
    // index에 document 추가 (post or put - 은 id가 있어야함)
    const response = await client.index({
        index: 'test-index',
        id: '2',
        // refresh 전체로 하는 거 - timeout이 있어서 이걸 해줘야 아래 get이 되나봄
        refresh: true,
        body: { foo: 'bar2' }
    });
    console.log('response=', response);
}

run().catch(e => {
    console.log('error=', e);
    process.exit(1);
})

 

 

 

2) index의 document 조회하기 

    // 정보 조회하기 get (read)
    const response2 = await client.get({
        index: 'test-index',
        id: '1',
    });
    console.log('response2=,', response2);

 

3) elasticsearch 에 있는 모든 index들을 조회하기 

    // 전체 indices 얻기
    const allIndices = await client.cat.indices({format: 'json'});
    console.log('all indices =', allIndices);

 

 

4) 특정 index의 모든 documents 조회하기

    // index 의 모든 documents 얻기
    const allDocuments = await client.search({
        index: 'test'
    })
    console.log('allDocuments =', allDocuments);
    console.log('allDocuments =', allDocuments.hits);
    console.log('allDocuments =', allDocuments.hits.hits);

 

 

5) 인덱스 생성하기 

    // index 생성하기
    const createIndex = await client.indices.create({
        index: 'create-index'
    })
    console.log('createIndex=', createIndex);

createIndex= {
  acknowledged: true,
  shards_acknowledged: true,
  index: 'create-index'
}

인덱스를 생성할 때 mappings (= 스키마) 해주기 

    await client.indices.create({
        index: 'test',
        body: {
            mappings: {
                dynamic: false, -> "true, false, stric" 옵션 있음
                properties: {
                    id: { type: 'keyword' },
                    title: { type: 'text' },
                    body: { type: 'text' },
                    answer_count: { type: 'integer' },
                    comment_count: { type: 'integer' },
                    creation_date: { type: 'date' },
                    tags: { type: 'keyword' }
                }
            }
        }
    })

위의 코드를 분석해보면 index명은 'test'로 만들고 body의 mappings 부분에 해당 index(db)의 스키마를 정해준다. 

dynamic: false => 정한 properties와 다른 데이터가 들어와도 괜찮음

dynamic: true => 정한 properties와 일치하는 데이터만 들어와야 함

dynamic: stric => 정한 properties와 엄격하게 일치해야 함. (그렇지 않으면 reject) 

 

elasticsearch의 type 중 'keyword'와 'text' 부분의 차이점에 대해 찾아보았는데 text는 역인덱스가 되고 형태소 분석 등이 이루어져 "검색" 용도로 적합하다. keyword의 경우 형태소 분석이 되지 않고 문자열 그대로 매칭되는 것만 유효하다. 예를 들면, "apples" 의 경우 keyword일 때 "lots of apples" 문자열 안의 apples과 매칭되지 않는다. "apples"와 동일한 키워드만 매칭된다. 

 

 

 

 

6) 인덱스 삭제하기

    // index 삭제하기
    const deleteResult = await client.indices.delete({
        index: 'create-index'
    })
    console.log('deletedResult=', deleteResult);

deletedResult= { acknowledged: true }

 

 

 

7) bulk insert - 대량의 json data를 elasticsearch에 입력하기 

먼저 split2 설치하기 

npm i --save-dev @types/split2

fs, path import 해오기  

import { Client } from '@elastic/elasticsearch';
import fs from 'fs';
import path from 'path';
import split from 'split2';

const client = new Client({
    node: 'http://localhost:9200'
})

const prepare = async() => {
    const doesItExist = await client.indices.exists({ index: 'test'});   // returns boolean
    console.log('doesItExist=',doesItExist);
    if (doesItExist) return;       // 해당 index가 존재한다면 return;

    await client.indices.create({  // 해당 index가 존재하지 않는다면 index create
        index: 'test',
        body: {
            mappings: {
                // strict : 아래 properties 에 맞지 않으면 거절됨
                dynamic: false,
                properties: {
                    id: { type: 'keyword' },
                    title: { type: 'text' },
                    body: { type: 'text' },
                    answer_count: { type: 'integer' },
                    comment_count: { type: 'integer' },
                    creation_date: { type: 'date' },
                    tags: { type: 'keyword' }
                }
            }
        }
    })
}

async function index() {
    const datasetPath = path.join(__dirname, '..', 'fixtures', 'test-bulk.ndjson');
    const datasource = fs.createReadStream(datasetPath).pipe(split());
    console.log('datasetPath=', datasetPath);
    console.log('datasource=', datasource);

    const result = await client.helpers.bulk({
        datasource,
        onDocument (doc) {
            return {
                index: { _index: 'test' }
            }
        }
    })
    console.log('result=', result);
}

prepare()
    .then(index)
    .catch(e => {
    console.log('error=', e);
    process.exit(1);
})

Kibana로 방금 입력한 내용 조회해보기 

bulk 로 들어간 데이터는 위에서 설정한 mappings와 일치하지 않지만 dynamic: false를 주어 상관없이 잘 입력되었다. 

 

 

document의 _id와 document data 안의 _id 값을 동일하고 세팅하고 싶을 때 

interface ITest {       // ITest type 선언
    id: string
    title: string
    body: string
    answer_count: number
    comment_count: number
    creation_date: string
    tags: string[]
}

async function index() {
    const datasetPath = path.join(__dirname, '..', 'fixtures', 'test-bulk.ndjson');
    const datasource = fs.createReadStream(datasetPath).pipe(split(JSON.parse));
    console.log('datasetPath=', datasetPath);
    console.log('datasource=', datasource);

    const result = await client.helpers.bulk<ITest>({  // 타입 지정
        datasource,
        onDocument (doc) {
            console.log('typeof doc=', typeof doc); // string -> JSON.parse를 넘기지 않는 경우
            console.log('typeof doc=', typeof doc); // object -> JSON.parse를 넘기는 경우

            return {
                index: { _index: 'test', _id: doc.id }   // _id: doc.id 추가
            }
        }
    })
    console.log('result=', result);
}

만약 <ITest> 타입을 bulk뒤에 적어주지 않으면 doc.id 부분에서 doc <- 에러가 난다. 

 

 

index함수에 onDrop() 함수를 추가하여 만약 데이터 타입이 달라 (strick mode) 들어가지 못하는 경우 해당 error를 내고 process.exit하도록 만들 수 있다. 

    const result = await client.helpers.bulk<ITest>({
        datasource,
        onDocument (doc) {
            console.log('typeof doc=', typeof doc); // string -> JSON.parse를 넘기지 않는 경우
            console.log('typeof doc=', typeof doc); // object -> JSON.parse를 넘기는 경우

            return {
                index: { _index: 'test', _id: doc.id }
            }
        },
        onDrop (doc) {   // 요기
            console.log(`can't index document ${doc.document.id}`, doc.error);
            process.exit(1);
        }
    })

 

bulk 를 상요할 때 다양한 bulk helpers (옵션) 이  존재한다. 필요에 따라 사용하면 유용할듯 하다. 

 

elasticsearch helpers 설명 및 사용법 포스팅 👇

https://blckchainetc.tistory.com/383

[Elasticsearch] Bulk helpers