[Elasticsearch] 연습

 

🔥 Elasticsearch 실험 🔥

 

1번 - split2가 왜 필요한지 궁금해서 .pipe(split(JSON.parse))를 빼봄

조건 : 

datasource = fs.createReadStream(datasetPath); 

data = test-bulk2.ndjson ( mappings와 맞는 스키마) 

mappings - dynamic : false 

allDocument =  [
  {
    _index: 'test',
    _type: '_doc',
    _id: 'BsLSK4AByTPDy5Ar5LWO',
    _score: 1,
    _source: { type: 'Buffer', data: [Array] }
  }
]

_source.data 가 아래처럼 나옴 

 

 

 

2번 - split2를 다시 넣어봄 .pipe(split(JSON.parse))

조건 : 

datasource = fs.createReadStream(datasetPath).pipe(split(JSON.parse)); 

data = test-bulk2.ndjson ( mappings와 맞는 스키마) 

mappings - dynamic : false 

=> error= SyntaxError: Unexpected end of JSON input 에러남 

 

 

 

3번 - dynamic: true 로 변경

조건 : 

datasource = fs.createReadStream(datasetPath).pipe(split(JSON.parse)); 

data = test-bulk2.ndjson ( mappings와 맞는 스키마 데이터 ) 

mappings - dynamic : false 

=> error= SyntaxError: Unexpected end of JSON input 에러남 

 

 

 

 

4번 - dynamic: false 로 다시 변경 / test-bulk.ndjson

조건 : 

datasource = fs.createReadStream(datasetPath).pipe(split(JSON.parse)); 

data = test-bulk.ndjson ( mappings와 맞는 스키마) 

mappings - dynamic : false 

=> 잘됨 

 

data 만 바꿨는데 왜지 !? 

 

 

 

5번 -  test-bulk2.ndjson / JSON.parse제거

조건 : 

datasource = fs.createReadStream(datasetPath).pipe(split()); 

data = test-bulk2.ndjson ( mappings와 맞는 스키마) 

mappings - dynamic : false 

=> error= DeserializationError: Unexpected end of JSON input

별렬 구조 문제? 에러가 나옴

근데 search를 해보면 데이터들은 모두 잘 들어가있음 -> 하지만 parse가 안되어 있어서 object가 아닌 string으로 처리됨 -  Id값은 다름 

 

 

 

 

6번 -  JSON.parse 추가

조건 : 

datasource = fs.createReadStream(datasetPath).pipe(split(JSON.parse)); 

data = test-bulk2.ndjson ( mappings와 맞는 스키마) 

mappings - dynamic : false 

=> error= SyntaxError: Unexpected end of JSON input

 

 

 

 

7번 -  새로운 ndjson 파일 생성 - test-bulk2.ndjson과 똑같은 데이터

조건 : 

datasource = fs.createReadStream(datasetPath).pipe(split(JSON.parse)); 

data = test-bulk2.ndjson ( mappings와 맞는 스키마) 

mappings - dynamic : false 

=> 성공 

같은 데이터인데 왜 이건 되었을까,,,

 

 

 

결론 :

- JSON.parse는 string - >object로 사용할 때 (doc.id ) 이렇게 쓸 때 사용하면 됨 

- dynamic 이 "strict" 인데 정한 mappings와 타입이 다르면 아래의 에러가 나온다. (true, false는 타입이 달라도 오류가 안난다.) 

can't index document undefined {
  type: 'strict_dynamic_mapping_exception',
  reason: 'mapping set to strict, dynamic introduction of [index] within [_doc] is not allowed'
}

 

 

 

 

8. search 해보기 

const run = async() => {
    const response = await client.search({
        index: 'test',
        body: {
            query: {
                match: { title : 'title' }
            }
        }
    })
    console.log('response =', response);
    console.log('response =', response.hits.hits);
}

"title" 에 "title"이 들어간 데이터 검색 -> id: 1, 3, 5, 6 이 나옴 

response = [
  {
    _index: 'test',
    _type: '_doc',
    _id: '1',
    _score: 0.49216813,
    _source: {
      id: '1',
      title: 'title',
      body: 'body',
      count: '1',
      date: '2022-04-15',
      tags: [Array]
    }
  },
  {
    _index: 'test',
    _type: '_doc',
    _id: '3',
    _score: 0.49216813,
    _source: {
      id: '3',
      title: 'title',
      body: 'body222',
      count: '100',
      date: '2022-01-11',
      tags: [Array]
    }
  },
  {
    _index: 'test',
    _type: '_doc',
    _id: '5',
    _score: 0.36680454,
    _source: {
      id: '5',
      title: 'title 5',
      body: 'body222',
      count: '100',
      date: '2022-04-20',
      tags: [Array]
    }
  },
  {
    _index: 'test',
    _type: '_doc',
    _id: '6',
    _score: 0.36680454,
    _source: {
      id: '6',
      title: 'title title4',
      body: 'body222',
      count: '100',
      date: '2022-04-30',
      tags: [Array]
    }
  }
]

 

 

 

 

 

 

bulk-indexer.ts 

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'});
    console.log('doesItExist=',doesItExist);
    if (doesItExist) return;

    await client.indices.create({
        index: 'test',
        body: {
            mappings: {
                dynamic: false, // strict : 아래 properties 에 맞지 않으면 거절됨
                properties: {
                    id: { type: 'keyword' },
                    title: { type: 'text' },
                    body: { type: 'text' },
                    count: { type: 'integer' },
                    date: { type: 'date' },
                    tags: { type: 'keyword' }
                }
            }
        }
    })
}

interface ITest {
    id: string
    title: string
    body: string
    count: number
    date: string
    tags: string[]
}

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

    const result = await client.helpers.bulk<ITest>({
        datasource,      // 필수값인 인자값, An array, async generator or a readable stream to index/create/update/delete
        onDocument (doc) { // 필수값인 인자값 / datasource의 각 document에 대입됨
            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) { // document가 indexed 되지 못할 때 그리고 retries 최대수를 넘겼을 때
            console.log(`can't index document ${doc.document.id}`, doc.error);
            process.exit(1);
        },
        refreshOnCompletion: true
    })
    console.log('result=', result);

    const allDocu = await client.search({
        index: 'test'
    });
    console.log('allDocument = ', allDocu);
    console.log('allDocument = ', allDocu.hits);
    console.log('allDocument = ', allDocu.hits.hits);
}

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

test-bulk2.ndjson

{"id": "1", "title": "title", "body": "body", "count": "1", "date": "2022-04-15", "tags": ["a", "b"] }
{"id": "2", "title": "title222", "body": "body222", "count": "100", "date": "2022-04-16", "tags": ["c", "d", "e"] }
{"id": "3", "title": "title", "body": "body222", "count": "100", "date": "2022-01-11", "tags": ["a"] }
{"id": "4", "title": "title4", "body": "body222", "count": "100", "date": "2022-04-11", "tags": ["b"] }
{"id": "5", "title": "title 5", "body": "body222", "count": "100", "date": "2022-04-20", "tags": ["c"] }
{"id": "6", "title": "title title4", "body": "body222", "count": "100", "date": "2022-04-30", "tags": ["a", "b", "c", "d", "e"] }