데이터 다루기

implementation 'com.fasterxml.jackson.core:jackson-databind:2.16.1' implementation 'co.elastic.clients:elasticsearch-java:8.11.4' implementation 'com.fasterxml.jackson.module:jackson-module-kotlin:2.16.1' implementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.16.1'

private fun getEsClient(): ElasticsearchClient { // Elasticsearch REST 클라이언트 빌더 생성 val restClientBuilder = RestClient.builder( HttpHost("localhost", 9200, "http") ) // 빌더를 사용하여 낮은 수준의 REST 클라이언트 인스턴스 생성 val lowLevelRestClient = restClientBuilder.build() // Jackson JSON 매퍼 구성. Java 8 날짜/시간 API 모듈 추가 및 몇 가지 옵션 설정 val mapper = jacksonMapperBuilder() .addModule(JavaTimeModule()) // Java 8 날짜/시간 모듈 추가 .configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false) // 날짜를 타임스탬프로 쓰지 않도록 설정 .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false) // 알 수 없는 속성에 대해 실패하지 않도록 설정 .build() // Jackson JSON 매퍼를 사용하여 REST 클라이언트 트랜스포트 설정 val transport = RestClientTransport(lowLevelRestClient, JacksonJsonpMapper(mapper)) // 최종적으로 Elasticsearch 클라이언트 인스턴스 반환 return ElasticsearchClient(transport) }

data class MyIndexClass( val fieldOne: String, val fieldTwo: Long, val fieldThree: ZonedDateTime ) data class MyPartialIndexClass( val fieldOne: String )

// 색인 private fun indexExample(client: ElasticsearchClient) { val indexResponse = client.index { builder -> builder .index("my-index") //색인 요청 함수 index .id("my-id-1") .routing("my-routing-1") .document(MyIndexClass("hello", 2L, ZonedDateTime.now(ZoneOffset.UTC))) } val result = indexResponse.result() println("result: $result") } // 조회 private fun getExample(client: ElasticsearchClient) { val request = GetRequest.Builder() .index("my-index") .id("my-id-1") .routing("my-routing-1") .build() client.get(request, MyIndexClass::class.java) .run { println("result: $this") } }

// Builder Pattern을 사용한 bulk API private fun bulkExampleOne(client: ElasticsearchClient) { val createOperation = CreateOperation.Builder<MyIndexClass>() .index("my-index") .id("my-id-2") .routing("my-routing-2") .document(MyIndexClass("world", 2L, ZonedDateTime.now(ZoneOffset.UTC))) .build() val indexOperation = IndexOperation.Builder<MyIndexClass>() .index("my-index") .id("my-id-3") .routing("my-routing-3") .document(MyIndexClass("world", 4L, ZonedDateTime.now(ZoneOffset.UTC))) .build() val updateAction = UpdateAction.Builder<MyIndexClass, MyPartialIndexClass>() .doc(MyPartialIndexClass("world updated")) .build() val updateOperation = UpdateOperation.Builder<MyIndexClass, MyPartialIndexClass>() .index("my-index") .id("my-id-1") .routing("my-routing-1") .action(updateAction) .build() val bulkOne = BulkOperation.Builder().create(createOperation).build() val bulkTwo = BulkOperation.Builder().index(indexOperation).build() val bulkThree = BulkOperation.Builder().update(updateOperation).build() val operations = listOf(bulkOne, bulkTwo, bulkThree) val response = client.bulk { it.operations(operations) } response.items().forEach { println("result: ${it.result()}, error: ${it.error()}") } } // Functional Call Pattern을 사용한 bulk API private fun bulkExampleTwo(client: ElasticsearchClient) { val bulkResponse = client.bulk { _0 -> _0 .operations { _1 -> _1 .index { _2: IndexOperation.Builder<MyIndexClass> -> _2 .index("my-index") .id("my-id-4") .routing("my-routing-4") .document(MyIndexClass("world", 4L, ZonedDateTime.now(ZoneOffset.UTC))) } } .operations { _1 -> _1 .update { _2: UpdateOperation.Builder<MyIndexClass, MyPartialIndexClass> -> _2 .index("my-index") .id("my-id-2") .routing("my-routing-2") .action { _3 -> _3 .doc(MyPartialIndexClass("world updated")) } } } } bulkResponse.items().forEach { println("result: ${it.result()}, error: ${it.error()}") } }

private fun searchExample(client: ElasticsearchClient) { val response = client.search({ builder -> builder .index("my-index") .from(0) .size(10) .query { query -> query .term { term -> term .field("fieldOne") .value { value -> value.stringValue("world") } } } }, MyIndexClass::class.java) response.hits().hits().forEach { println("result: $it") } }

// 대량(bulk) 작업의 생명 주기를 관리하는 리스너 class BulkIngestListener<Context> : BulkListener<Context> { override fun beforeBulk(executionId: Long, request: BulkRequest?, contexts: MutableList<Context>?) { println("[${LocalDateTime.now()}] beforeBulk - executionId: $executionId") } override fun afterBulk( executionId: Long, request: BulkRequest?, contexts: MutableList<Context>?, response: BulkResponse? ) { println( "[${LocalDateTime.now()}] " + "afterBulk - executionId: $executionId, " + "itemSize: [${response?.items()?.size}], " + "took : ${response?.took()}, " + "errors: ${response?.errors()}" ) } // 작업이 완료되기전 혹은 오류가 발생한 후 호출하는 메서드 override fun afterBulk( executionId: Long, request: BulkRequest?, contexts: MutableList<Context>?, failure: Throwable? ) { System.err.println(failure) } } private fun bulkIngesterExample(client: ElasticsearchClient) { val listener = BulkIngestListener<String>() val ingester = BulkIngester.of<String> { it.client(client) // Elasticsearch 클라이언트를 설정합니다. .maxOperations(200) // 대량 작업에서 처리할 최대 작업 수를 설정합니다. 이 값이 너무 낮으면 오버헤드가 증가하고, 너무 높으면 메모리 문제가 발생할 수 있습니다 .maxConcurrentRequests(1) // 동시에 처리할 최대 요청 수를 설정합니다. 이 값은 Elasticsearch 클러스터의 용량과 원하는 병렬 수준에 따라 설정해야 합니다 .maxSize(5242880L) // 대량 요청의 최대 크기를 바이트 단위로 설정합니다. 이 값은 사용 가능한 메모리와 문서의 크기에 따라 설정해야 합니다 .flushInterval(5L, TimeUnit.SECONDS) // 대량 요청이 실행되는 간격을 설정합니다. 이 간격은 작업 수나 요청 크기에 상관없이 적용됩니다 .listener(listener) // 대량 작업의 생명주기를 관리하는 리스너를 설정합니다. } for (number in 0L until 1100L) { val bulkOperation = BulkOperation.of { bulkBuilder -> bulkBuilder.index { indexOpBuilder: IndexOperation.Builder<MyIndexClass> -> indexOpBuilder .index("my-index") .id("my-id-$number") .routing("my-routing-$number") .document(MyIndexClass("world", number, ZonedDateTime.now(ZoneOffset.UTC))) } } ingester.add(bulkOperation) } println("[${LocalDateTime.now()}] sleep 10 seconds ...") Thread.sleep(10000L) for (number in 1100L until 1200L) { val bulkOperation = BulkOperation.of { bulkBuilder -> bulkBuilder.index { indexOpBuilder: IndexOperation.Builder<MyIndexClass> -> indexOpBuilder .index("my-index") .id("my-id-$number") .routing("my-routing-$number") .document(MyIndexClass("world", number, ZonedDateTime.now(ZoneOffset.UTC))) } } ingester.add(bulkOperation) } println("[${LocalDateTime.now()}] sleep 10 seconds ...") Thread.sleep(10000L) ingester.close() }