쏘니의 개발블로그

서브쿼리(Subquery)란 sql내에 다른 sql이 포함되어있는 포함되어있는 쿼리를 의미합니다.

스칼라 서브쿼리(Scalar Subquery)는 주로 SELECT 절이나 다른 쿼리의 일부로서 단일 값을 반환하는 서브쿼리를 말합니다. 스칼라 서브쿼리는 반드시 단일 값(하나의 행과 하나의 열)을 반환해야 하며, 이를 통해 주 쿼리에서 계산이나 조건 비교 등의 작업에 사용할 수 있습니다.

WHERE 절이나 HAVING 절에 사용되는 서브쿼리는 서브쿼리지만, 일반적으로 “스칼라 서브쿼리”라고 지칭하지 않습니다. 보통 WHERE 절에 사용되는 서브쿼리는 존재 여부를 확인하는 서브쿼리(Existential Subquery) 또는 다중 행 서브쿼리(Multi-row Subquery)로 불릴 수 있습니다.

스칼라 서브쿼리의 사용 예

1. SELECT 절에서 사용

스칼라 서브쿼리는 주로 SELECT 절에서 하나의 값을 계산하거나 가져오는 데 사용됩니다. 예를 들어, 각 직원의 부서 평균 급여를 가져오는 경우

SELECT employee_id, first_name, last_name, salary,
       (SELECT AVG(salary) FROM employees WHERE department_id = e.department_id) AS avg_department_salary
FROM employees e;

위 쿼리의 수행 순서는 메인 쿼리 -> 서브 쿼리 순으로 동작합니다.

서브쿼리는 결과 건수만큼 반복 수행되기 때문에 조회되는 데이터의 갯수가 적다면 크게 문제될 것은 없지만,
조회 되는 데이터의 갯수가 많아지면 많아질 수록 성능이 현저히 떨어지는 모습을 볼 수 있습니다.

또한 스칼라 서브쿼리 사용시 서브쿼리의 결과값이 같은 경우가 많거나, 매번 동일한 결과값을 반환한다면, 스칼라 서브쿼리의 캐싱 효과를 통해 성능을 높일 수 있으나, 서브쿼리에서 사용되는 조건의 종류가 많아지거나 혹은 조건 데이터가 지속적으로 바뀔경우 캐싱의 효율성이 떨어지기 때문에 이전보다 성능이 떨어질 수 있습니다.

서브쿼리는 매번 변하는 결과값을 모든행마다 조회해야한다면 가능하다면 from 절에서 join 하는 방식으로 수정해야 한다.

SELECT e.employee_id, e.first_name, e.last_name, e.salary, d.avg_department_salary
FROM employees e
LEFT JOIN (
    SELECT department_id, AVG(salary) AS avg_department_salary
    FROM employees
    GROUP BY department_id
) d ON e.department_id = d.department_id;

카프카 오류

Synchronous auto-commit of offsets {topic-2=OffsetAndMetadata{offset=96, leaderEpoch=0, metadata=''}} failed: Offset commit cannot be completed since the consumer is not part of an active group for auto partition assignment; it is likely that the consumer was kicked out of the group.

위와 같은 오류가 발생해서 찾아보니 다음과 같다.


Kafka에서 CommitFailedException이 나오는 유형은 아래와 같다.

consumer 로직의 처리시간이 max.poll.interval.ms보다 클 경우 리밸런싱으로 인한 컨슈머 그룹에서 제외되었을 경우

session.timeout.ms시간동안 heartbeat가 오지 않았을 경우 리밸런싱으로 인한 컨슈머 그룹에서 제외되었을 경우

1. consumer 로직의 처리시간이 max.poll.interval.ms보다 클 경우 리밸런싱으로 인한 컨슈머 그룹에서 제외되었을 경우

ConsumerConfig 설정 방법 
아래 방식대로 consume 로직 처리시간 변경가능.

@Bean
public ConsumerFactory<String, JsonSerializable> consumerFactory() {
	...
    props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 5000); // 5초로 설정
    ...
}

위의 로직이 완료되고 commit될 때 CommitFailedException이 발생한다. 

max.poll.interval.ms 는 기본 5분으로 알고 있다.

This error handler cannot process 'org.apache.kafka.clients.consumer.CommitFailedException's; no record information is available

2. session.timeout.ms시간동안 heartbean가 오지 않았을 경우 리밸런싱으로 인한 컨슈머 그룹에서 제외되었을 경우

heartbeat 메시지는 consumer에서 3초(기본값)에 한번씩 주기적으로 broker에 날린다.
메시지 내용을 확인하려면 logger에 아래 내용을 추가한다.

logging:
	level:
    	org.apache.kafka.clients.consumer.internals.AbstractCoordinator: debug

subscriber의 로직에 breakpoint를 걸면 heartbeat 메시지가 날라가지 않으므로 10초 후에 리밸런싱이 되고 컨슈머 그룹에서 제외된다.
그때 로직이 수행되고 커밋하려면 CommitFailedException이 발생한다.

https://logz.io/blog/logstash-grok/

https://www.elastic.co/kr/blog/a-practical-introduction-to-logstash

grok debugger

https://grokdebug.herokuapp.com/

acks 란

acks는 acknowledgments의 약자로 사전에서 찾아 보면 "승인", 확인
프로듀서가 메시지를 보내고 그 메시지를 카프카가 잘 받았는지 확인을 할 것인지 또는 확인을 하지 않을 것인지를 결정하는 옵션

acks 옵션

acks=0 

단순 메트릭 정보와 같이 메세지 손실이 어느정도 눈감아지는 상황인 경우 사용.

acks=1

producer 가 kafka 에 데이터 전송 -> leader broker 는 쓰기요청에 대해 producer 에게 respond 를 보낸다. 그리고 데이터를 topic 에 쓴다.

leader 에게 응답받지 못한 producer 는 다시 쓰기 요청을 재시도 하고, leader 의 replica 들도 제대로 메시지를 받았는지 확인할 수 없다.

replica 들이 제대로 leader 의 topic/partition 을 제대로 복제 못한 상태에서 leader 가 다운되면 메시지가 손실 될 수 있다.

acks = all

leader 뿐 아니라 replicas 들도 ack 를 보내야 한다.

그만큼 지연시간은 늘어나지만 데이터 손실이 되지 않는다. 하나의 데이터도 손실이 용납되지 않는 경우 해당 옵션 사용 해야 한다.

acks가 all 인 경우 min.insync.replicas 옵션을 고려해야 한다.

min.insync.replicas=2 는 적어도 2개의 브로커가 데이터를 받았다는 ack응답을 보내야 한다는 뜻.

replication factor=3, min.insync.replicas=2, acks=all 인 경우 -> 모든 브로커에게 ack를 받지 못해도 min.insync.replicas 가 2이기때문에 1개가 장애나도 프로세스는 유지된다. 2개가 문제인 경우 producer 는 NOT_ENOUGH_REPLICAS 라는 예외를 받는다.

retry

NOT_ENOUGH_REPLICAS 과 같은 일시적 장애가 생긴경우 다시 시도하는 횟수.

kafka 2.1 버젼 이상부터는 retries 가 2147483647 번으로 기본값 설정됨.

retry.backoff.ms 옵션의 기본값은 100ms 이다.

delivery.timeout.ms

위 옵션이 2분인 경우 producer 는 acks 를 받지 못한 메시지에 대해 2분동안 요청 retry 를 한다.

위 시간동안 ack 를 받지 못하면 fail 이다.

retries 가 일어나는 동안 메시지의 순서는 보장되지 않는다.

key 기반 ordering 을 원하는 경우 문제가 발생한다.

이 경우 product request 가 병렬적으로 실행되도록 조절하는 옵션인 

max.in.flight.requests.per.connection 의 값을 조절하면 된다.

max.in.flight.requests.per.connection 

Default: 5

순서를 보장하기 위해서는 1로 설정해야하지만, 처리량은 낮아질 수 있다.

acks 를 보내는 과정에서 네트워크 에러가 발생하는 경우 request 가 중복이 되는 경우가 발생할수 있다.

kafka 0.11 버젼부터는 Idempotent request 라는 설정이 가능하다.

produce request 에 id 가 부여되기 때문에 broker 단에서 중복되는 reqeust 를 알아채고 중복을 방지한다.

Batching

max.in.flight.requests.per.connection = 5 의 경우 동시에 5건의 메시지가 개별적으로 전달된다. 

linger.ms -> produce 가 batch 를 보내기 전에 기다리는 시간 (ms)

값을 올릴수록 해당 초까지 기다렸다가 같이 보낸다.하지만 해당 시간 전에 batch.size 만큼 메시지가 차면 바로 배치로 보낸다.

batch.size 

하나의 배치안 에 넣을 수 있는 최대 바이트 수.

배치 사이즈를 늘리면 요청을 보낼때 압축률, 처리량, 효율성에 이점을 볼 수 있다. 

배치 사이즈보다 더 큰 사이즈의 메시지의 경우 배치로 처리되지 못한다.

하나의 배치는 파티션별로 할당되기 때문에 너무 높은 값으로 정하면 메모리 부족이 생길 수 있다.

memory

broker가 메시지를 처리하는 속도보다 producer 가 더 빠르게 메시지를 보내는 경우 해당 레코드는 잠시 producer 메모리에 buffer 된다.

buffer.memory 

기본값은 32MB, send buffer 의 사이즈다.

이 버퍼는 시간이 지남에 따라 계속 차고, 브로커에 더 빠르게 메시지를 보낼 수 있게 되면 다시 내려간다.

버퍼가 가득 차면 send 메소드는 block 된다. 데이터를 못보내고 그냥 대기하게 된다.

max.block.ms

send() 메소드가 예외를 던지기까지 block 되는 시간이다.

예외를 던지는 경우 

1. producer 버퍼 메모리가 꽉 찬 경우.

2. broker 가 전혀 새로운 데이터를 받지 못한 경우.

3. max.block.ms 시간이 다 지난 경우.

https://www.popit.kr/kafka-%EC%9A%B4%EC%98%81%EC%9E%90%EA%B0%80-%EB%A7%90%ED%95%98%EB%8A%94-producer-acks/

https://4betterme.tistory.com/168

query_string 쿼리는 여러 조건을 조합하기에는 용이한 문법이지만 옵션이 한정되어 있다.

여러 쿼리를 조합하기 위해서는 상위에 bool 쿼리를 사용하고 그 안에 다른 쿼리들을 넣는 식으로 사용 가능하다.

bool 쿼리는 4개의 인자를 가지고 있고, 그 인자 안에 다른 쿼리들을 배열로 넣는 방식으로 동작한다.

• must : 쿼리가 참인 도큐먼트들을 검색
• must_not : 쿼리가 거짓인 도큐먼트들을 검색
• should : 검색 결과 중 이 쿼리에 해당하는 도큐먼트의 점수를 높인다
• filter : 쿼리가 참인 도큐먼트를 검색하지만 스코어를 계산하지 않는다. must 보다 검색 속도가 빠르고 캐싱이 가능.

GET <인덱스명>/_search
{
  "query": {
    "bool": {
      "must": [
        { <쿼리> }, …
      ],
      "must_not": [
        { <쿼리> }, …
      ],
      "should": [
        { <쿼리> }, …
      ],
      "filter": [
        { <쿼리> }, …
      ]
    }
  }
}

and 조건으로 검색하려면

must 배열안에 여러개의 조건을 추가해야 한다.

예를 들어 match 해당 기간동안 logDate의 로그에 어떤 text 를 가진 데이터 조회를 할 경우

GET 인덱스/_search
{
  "query": { 
    "bool": { 
      "must": [
        {
          "match": {
              "text": "텍스트"
          }
        },
        { 
          "range": { 
             "logDate": {
                  "gte": "2021-06-29 11:00:00.000",
                  "lte": "2021-06-29 12:00:00.000",
                  "format": "yyyy-MM-dd HH:mm:ss.SSS",
                  "time_zone": "+09:00"
             }
          }
        }
      ]
    }
  }
}

or 조건으로 검색하려면 must : [{match :"A B"}] 와 같이 검색하자.

참고문헌

> https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl-bool-query.html [bool 쿼리 공식문서]

> https://esbook.kimjmin.net/05-search/5.2-bool [블로그 Elastic 가이드북]

like 검색도 in 처럼 여러개 검색이 가능하다.

in 절 처럼 like 여러개 검색하고 싶은 경우

where 컬럼 REGEXP ('문자열1|문자열2|문자열3')

1. ORDER BY FIELD() 를 사용하여 '자전거'가 첫번째, '비행기'가 두번째, 그외 나머지는 오름차순으로 정렬하기

SELECT 
	product_name
FROM
    COIN
ORDER BY
    FIELD(product_name, '자전거', '비행기') DESC,
    product_name ASC

- FIELD 함수의 첫번째 파라미터에 정렬할 컬럼을 넣고, 그 이후 파라미터엔 우선적으로 정렬할 값들을 역순으로 넣어준다.

그리고 FIELD 함수를 DESC정렬 시켜주면 된다.

- 원리는 FIELD 함수를 사용하면 첫번째 파라미터의 값과 같은 값이 그 이후 파라미터의 몇번째에 있는지 index값을 반환한다.

즉 이후 파라미터의 첫번째에 있는 '자전거'는 1이 반환되고, 두번째에 있는 '비행기'는 2가 반환이 된다.

같은 값이 없는 나머지는 0이 반환된다.

이것을 DESC정렬하여 우선적으로 정렬되게 한다. 

2. order by 에 case 문을 사용

SELECT
	*
FROM 
	table A
ORDER BY 
	CASE WHEN A.product_name = '자전거' THEN 1 
	ELSE
		2
	END

똑같은 원리다. 상위에 정렬 시키고 싶은 내용들을 case 문의 상수값을 작은 숫자로 반환시켜주면 된다.

몽고디비 조회 속도가 너무 느려져서 인덱스를 추가했다.

추가하는데 엄청난 시간이 들지 예상하지 못해, 생성문을 실행했는데, 이게 실행이 되지 않아 당연히 클라이언트에서 실행중 수행시간이 길어져 프로세스가 꺼진줄 알고 그냥 뒀다.

robo 3T 라는 툴에서는 프로세스가 15초 이상 실행되어 실행되지 않는다~ 뭐 이런 식의 문구가 떠서 끄면 작업이 끝난줄 알았다.

그런데 이게 최초에 foreground 로 돌리던게 돌아갔던건지 background 로 돌리던게 돌아가는건지 몽기DB 쓰기 수행작업이 lock 이 걸렸습다.

백그라운드로 돌렸어도 전체에 락이 걸렸을거 같은데 확실치 않다.

인덱스 생성 명령어를 수행한지 2시간이 지나도 lock 이 안풀려있엇어서 서비스 병목 현상때문에 알게 되었다. 

나중이 되서야 인덱스 생성되는 프로세스를 끌 수 있다는걸 그 뒤에 알았다 ㅠㅠ.

안일하게 생성명령어만 돌리고 생성안되었다고 방치하지만 않았어도 이슈가 발생하지 않았는데 말이다.

최초에 금방 생성될 줄 알고 실수로 포그라운드에서 인덱스 생성을 날렸어서,

클라이언트에서 중지됐다는 alert 문구와 별개로 계속 백그라운드에서 실행중이였나 보다.

백그라운드로 돌려도 그런 형상이 나오는지는 잘 모르겠다.

해서 현재 만들어지고 있는 인덱스 생성 오퍼레이션을 취소해줘야 한다.

아래와 같은 방법이 있는지 알았으면.....

killIndexMaking = function(a) {
    currOp = db.currentOp({"op": "command", 'query.createIndexes':'인덱스를 건 컬렉션명'})['inprog']
    //print(currOp) // 만들고있는 인덱스가 맞는지 check
    print("Killing opId: " + currOp[0].opid);
    //db.killOp(currOp[0].opid); // 죽인다.
};
 
killIndexMaking(1)

db.currentOp명령어로 'query.createIndexes'컬럼을 필터로 걸어 해당 컬렉션에 걸리고있는 인덱스 목록을 가져온다. 위에서는 인덱스를 하나만 만들고 있다고 가정하고 첫번째(currOp[0].opid)를 죽인다. 

일단 실행전에 인덱스 생성 리스트를 확인해본다.

리스트를 확인해보는 방법은 여러개가 있다.

db.currentOp(true).inprog.forEach(function(op){ if(op.msg!==undefined) print(op.msg) })

db
  .currentOp({"command.createIndexes": { $exists : true } })
  .inprog
  .forEach(function(op){ print(op.msg) })

db.currentOp({ 
    'msg' :{ $exists: true },
    'command': { $exists: true },
    $or: [ 
        { 'command.createIndexes': { $exists: true } }, 
        { 'command.reIndex': { $exists: true } }
    ]
}).inprog.forEach(function(op) { 
    print(op.msg); 
});

op 를 찾아서 해당 오퍼레이션을 지워줘야 하는데 문서를 보다 보니

Use the dropIndexes command or its shell helpers dropIndex() or dropIndexes() to terminate an in-progress index build. See Abort In-Progress Index Builds for more information.
Do not use killOp to terminate an in-progress index builds in replica sets or sharded clusters.

이런 말이 나왔다. 

killOp 를 쓰지말고 dropIndex() 를 사용해라. 혹시나 이런 일이 생길 경우에 dropIndex 를 써봐야겠다.

docs.mongodb.com/manual/core/index-creation/ [공식문서]
qkqhxla1.tistory.com/998 [인덱스 생성 취소 스크립트]
https://stackoverflow.com/questions/22309268/mongodb-status-of-index-creation-job [인덱스 생성 중인 상태보기]