최근 근무를 하면서나 개인적으로 사이드 프로젝트를 진행할 때도 Python 기반의 DataOps 업무를 주로 진행했기에, SQL 관련공부에 소홀해졌던 것 같다. 물론 데이터를 전처리하거나 데이터마트를 생성할 때, DW 상에서 SQL 쿼리를 꽤 작성했지만, 유사한 구조를 지속적으로 사용하다보니 이전에 자격증을 공부할 때 학습한 스킬이나, 이미 사용하던 것들도 점점 개념이 잊혀져 간 것 같아, 유용하게 사용한 SQL function에 대해 복기해보려고 합니다 🔥

With문

With문을 사용하는 이유에 대해서 물어본다면, 가장 먼저 떠오르는 것은 ‘가독성’이다. 당장 ‘왜 사용했지?’ 라는 물음에, 기억을 더듬어보면 2개 이상의 서브쿼리(Subquery) 개수로 인한 가독성 이슈를 해결하고 싶어 활용했던 이유가 가장 컸던 것 같다. 개념을 설명하기 위해, 특정한 상황을 가정해봅시다 :)

예를 들어, 직원 테이블에서 부서별 평균 급여를 계산한 후, 각 직원의 급여가 해당 부서 평균보다 높은 직원을 조회하는 서브쿼리가 포함된 질의문이 있다고 해봅시다.

SELECT employee_id, salary, department_id
FROM employees e
WHERE salary > (
    SELECT AVG(salary)
    FROM employees
    WHERE department_id = e.department_id
);

위 쿼리에 With문을 적용하여 변환해보면 아래와 같을 것입니다 :)

WITH DeptAvgSalaries AS (
    SELECT department_id, AVG(salary) AS avg_salary
    FROM employees
    GROUP BY department_id
)
SELECT e.employee_id, e.salary, e.department_id
FROM employees e
JOIN DeptAvgSalaries d
ON e.department_id = d.department_id
WHERE e.salary > d.avg_salary;

이런식으로, 일명 CTE(Common Table Expression)를 사용하여 쿼리를 작성하면 복잡한 쿼리를 더 읽기 쉽게 만들고, 무엇보다 쿼리의 각 부분을 특정 이름으로 재사용할 수 있게 해줍니다.

특히 서브쿼리를 2개 이상 사용했을 때, With문을 적용한다면 이러한 재사용성의 장점이 더욱 부각됩니다. 아래는 부서별 평균 급여를 계산한 후, 이 데이터를 두 번 참조하여 각각 급여가 평균보다 높은 직원과 낮은 직원을 조회하는 예제입니다.

-- 서브쿼리로 평균보다 높은 직원 조회
SELECT e.employee_id, e.department_id, e.salary,
       (SELECT AVG(salary)
        FROM employees
        WHERE department_id = e.department_id) AS avg_salary
FROM employees e
WHERE e.salary > (SELECT AVG(salary)
                  FROM employees
                  WHERE department_id = e.department_id)

UNION ALL

-- 서브쿼리로 평균보다 낮은 직원 조회
SELECT e.employee_id, e.department_id, e.salary,
       (SELECT AVG(salary)
        FROM employees
        WHERE department_id = e.department_id) AS avg_salary
FROM employees e
WHERE e.salary < (SELECT AVG(salary)
                  FROM employees
                  WHERE department_id = e.department_id);

이를 아래와 같이 With문을 적용하여, 사전에 정의한 DeptAvgSalaries (쿼리 실행 시 메모리 상에 생성된 가상의 테이블)를 재사용한다면, 쿼리 실행 횟수를 2회에서 1회로 줄일 수 있기 때문에 효율성(쿼리 성능)을 높일 것이라고 쉽게 예상할 수 있을 것입니다 🙂

WITH DeptAvgSalaries AS (
    SELECT department_id, AVG(salary) AS avg_salary
    FROM employees
    GROUP BY department_id
)
-- 평균보다 높은 직원 조회
SELECT e.employee_id, e.department_id, e.salary, d.avg_salary
FROM employees e
JOIN DeptAvgSalaries d ON e.department_id = d.department_id
WHERE e.salary > d.avg_salary

UNION ALL

-- 평균보다 낮은 직원 조회
SELECT e.employee_id, e.department_id, e.salary, d.avg_salary
FROM employees e
JOIN DeptAvgSalaries d ON e.department_id = d.department_id
WHERE e.salary < d.avg_salary;

하지만 여기서 중요한 점은 With문의 동작 방식에 따라 효율성이 높아질 수도, 낮아질 수도 있다는 점입니다!

일단, 앞서 설명했던 흐름에 따라 우리가 예상하는 With문의 동작 방식은 Materialize 일 것입니다. 해당 방식은 With문을 통해 일종의 임시 테이블을 생성하고, 서브쿼리를 매번 실행하는 대신, 사전에 생성한 임시 테이블을 재사용하는 경우입니다.

물론 옵티마이저의 최적화 과정에 따라 달라질 수 있겠지만, 경험 상 일반적으로 대부분의 DBMS 상에서의 With문의 동작은 Inline View 방식으로 처리됩니다. 해당 방식은 말그대로 View 와 같이 쿼리 그 자체만 저장되기 때문에, With문을 통해 정의한 테이블을 조회할 때마다 쿼리를 실행합니다. 따라서, 실제로는 서브쿼리를 사용했을 때와 동일한 성능을 보일 것입니다. 대신, 가독성을 높일 수 있다는 점은 가져갈 수 있겠죠?

그렇다면, Materialize 방식을 채택한다면 늘 성능 개선을 이끌 수 있을까요? ** 정답은 예상하셨다시피 **'NO' 입니다. 아래의 2가지 상황에서는 위 동작방식의 With문의 적용을 지양해야 할 수 있습니다 😅

1. 결과 rows가 매우 클 경우: Materialize 방식의 쿼리 결과가 매우 큰 데이터셋을 포함하게 되면, 임시 테이블을 메모리에 생성하고 유지하는 데 드는 자원이 크게 증가하여 시스템의 부하를 가져올 수 있습니다. 상식적으로 임시 테이블을 생성(CREATE)하고, 쿼리가 끝난 후 임시 테이블을 삭제(DROP)하는 과정을 거치기 때문에, rows가 너무 많을 경우에는 이 과정 자체에서 시스템의 부하가 발생할 것입니다. 이때는 CTAS로 임시테이블을 생성하여, 인덱스나 파티셔닝을 적용하는 등의 별도의 최적화 과정을 고려하는 것이 좋을 것으로 예상됩니다 :)

2. 필터링이나 조건이 변경될 때: Materialize 방식으로 임시 테이블을 생성하면, CTE가 한 번 실행된 후 결과를 재사용합니다. 하지만 각 참조 시마다 조건이 달라질 경우, 그리고 심지어 rows가 많을 경우에는 임시 테이블이 효율적이지 않고 원하는 필터링이 잘 이루어지지 않을 확률이 높기 때문에 불필요한 데이터를 읽게 될 수 있습니다.

그렇다면, 언제 With문을 Materialize 방식으로 처리하는 것이 좋을까요?

*I/O 비용이 크지만, 결과 dataset이 작을 경우 매우 유용하게 사용할 수 있습니다! 🎉 *

예를 들어, 결과를 도출하는 과정에서 (예: join문 중첩) 복잡한 연산이 진행될 경우, 단 한번의 비용소모로 재사용의 이점을 극대화할 수 있고, 결과 rows가 작기 때문에 디스크나 메모리에 부하를 주지 않다는 점도 장점으로 작용할 것입니다.

TIP💡

앞서 언급했듯이 대부분의 DBMS에서는 기본적으로 Inline View를 채택하기 때문에 (물론, 옵티마이저의 최적화 과정에 따라 다를 수 있음!), Materialize 방식을 강제하기 위해서는 각 DBMS 별로 최척화 방법에 대해 이해하고, 이에 따른 대처를 달리 해야 합니다. 예를 들어, Oracle에서는 /*+ MATERIALIZE */를, MySQL에서는 /*+ NO_MERGE(cte) */ 와 같은 방식으로 옵티마이저에게 힌트를 제시해야 합니다. 따라서, 쿼리 성능 개선을 위해 With문을 적용하려고 한다면, 먼저 사용하는 DBMS이 With문을 어떻게 처리하는지에 대한 이해가 선행되어야 합니다 :)

Window 함수

Window 함수는 쿼리 내에서 특정 레코드 집합을 기준으로 데이터를 처리할 수 있게 하는 SQL 함수입니다. 일반적인 집계 함수와 달리, 결과를 그룹으로 묶지 않고도 각 행에 대해 연산을 수행할 수 있기 때문에, 유용하게 사용할 상황이 꽤 발생합니다. 이번 포스팅에서는 대표적인 윈도우 함수인 RANK(), ROW_NUMBER(), LAG(), LEAD()에 대해서 소개하겠습니다.

RANK(): 순위 계산

RANK() 함수는 특정 기준에 따라 데이터의 순위를 계산합니다. 동일한 값에 대해 같은 순위를 부여하고, 동일한 순위 이후에는 건너뛰는 순위가 적용됩니다. 예를 들어, 1위가 여러 개 있으면 다음 순위는 2위가 아닌 3위가 됩니다.

예시 쿼리

아래 쿼리는 직원의 급여를 기준으로 내림차순으로 정렬하여 순위를 계산합니다. 또한, 동일한 급여를 가진 직원들은 같은 순위를 부여받고, 다음 순위는 건너뛰게 될 것입니다.

SELECT employee_id, salary, 
       RANK() OVER (ORDER BY salary DESC) AS salary_rank
FROM employees;

쿼리 실행 결과

앞서 설명한 동작 원리에 따라, 급여가 10,000인 직원들은 1위를 공유하고, 그 다음 순위는 2가 아니라 3이 됩니다.

ROW_NUMBER(): 행 번호 반환

ROW_NUMBER() 함수는 특정 기준에 따라 각 행에 고유한 번호를 부여합니다. 동일한 값에 대해서도 중복 없이 번호가 매겨지며, 데이터의 순서에 따라 고유 번호가 부여됩니다.

예시 쿼리

아래 쿼리는 직원의 급여를 내림차순으로 정렬하고, 각 직원에게 고유한 행 번호를 부여합니다. 또한, 동일한 급여 값이라도 번호는 중복되지 않습니다.

SELECT employee_id, salary, 
       ROW_NUMBER() OVER (ORDER BY salary DESC) AS row_num
FROM employees;

쿼리 실행 결과

결과를 보면, ROW_NUMBER()는 순위와 달리 동일한 값에 대해 중복된 번호를 부여하지 않고, 고유한 번호를 차례대로 부여하는 것을 알 수 있습니다.

UPSERT

UPSERT는 Update + Insert를 합친 데이터 업데이트 방식입니다. 이름에서도 알 수 있듯이, UPSERT는 중복되는 값이 없다면 삽입(Insert)을 하고, 중복되는 값이 있다면 최신화(Update)를 하는 쿼리를 뜻합니다. 데이터 엔지니어로서 ROW_NUMBER()를 처음 접했던 때는 UPSERT 방식을 DW 상에서 구현하려고 시도했을 때였던 것 같아, ROW_NUMBER()에 대한 이해를 돕기 위해 추가로 설명해보려고 합니다 🔥

보통의 데이터웨어하우스 솔루션에서는 나름의 방식으로 UPSERT 방식을 지원하지만 (예: BigQuery의 merge into 문), 일단 이를 활용하지 않고 Incremental Update를 한다고 가정해봅시다 🫡 (참고로 Full refresh를 하는 경우는, 어차피 모두 재호출하여 적재하기 때문에 상관없을 것)

예를 들어, 최근 7일 간의 날씨 데이터에 대해 refresh하는 실행문을 작성해야하는 상황을 생각해본다면 아래와 같은 고려사항이 있을 것입니다.

보통의 RDBMS의 경우에는 DATE에 PK가 걸려있기 때문에 중복이 일어나지 않을 수 있지만, 데이터 웨어하우스 솔루션은 빅데이터에서의 성능을 보장하기 위해 일반적으로 PK를 지원하지 않기 때문에, 오늘 호출하고 내일 호출하면 데이터가 겹칩니다. 추가로, 같은 날짜에 대한 데이터라도 최근에 호출한 데이터의 신뢰도가 높기 때문에, 비교적 최근 데이터로 refresh 하려는 수요 또한 존재할 것입니다.

따라서, 이를 해결하기 위해서는 DATE가 동일한 것들끼리 grouping을 하고, created_date(DATE)가 큰 것부터 역순으로 sorting한 뒤, 일련번호를 붙여서 가장 최근 값만 채택을 해야합니다. 이때, ROW_NUMBER() 함수를 적용하여 아래와 같이 처리하면 됩니다.

INSERT INTO weather_daily_table
SELECT date, temp, min_temp, max_temp, created_date
FROM (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY date ORDER BY created_date DESC) seq
    FROM t
)
WHERE seq = 1

위와 같이 처리하면, 아래 사진과 같은 과정을 통해 Primary Key Uniqueness 보장하면서 신규 데이터로 refresh할 수 있습니다.

LAG(): 이전 행의 값 참조

LAG() *함수는 *현재 행 이전의 행 값을 가져옵니다. 주로 시계열 데이터나 이전 값과 비교가 필요한 상황에서 사용됩니다. 기본적으로 이전 행을 참조하지만, 참조할 행의 간격(몇 번째 이전 행)을 지정할 수 있습니다.

예시 쿼리

아래 쿼리는 직원의 고용일(hire_date)을 기준으로 이전 직원의 급여를 가져옵니다. 또한, 이전 행이 없을 경우 기본값으로 0을 반환합니다.

SELECT employee_id, salary,
       LAG(salary, 1, 0) OVER (ORDER BY hire_date) AS prev_salary
FROM employees;

쿼리 실행 결과

첫 번째 행의 이전 행이 없기 때문에 prev_salary는 0을 반환하고, 이후 행들은 각각 이전 직원의 급여를 참조합니다.

추가로, 이해를 돕기 위해 주식 가격 데이터에서 각 주식의 전일 대비 가격 변동을 계산해야 하는 상황을 가정한다면, *LAG() *함수를 활용하여 아래와 같은 쿼리를 작성할 수 있을 것입니다.

SELECT stock_id, date, price, 
       LAG(price, 1) OVER (PARTITION BY stock_id ORDER BY date) AS prev_price,
       price - LAG(price, 1) OVER (PARTITION BY stock_id ORDER BY date) AS price_change
FROM stock_prices;

LEAD(): 다음 행의 값 참조

LEAD() 함수는 현재 행 이후의 값을 가져옵니다. 주로 시계열 데이터나 다음 값과 비교가 필요한 상황에서 사용됩니다. LAG()와 반대로 다음 행을 참조하며, 몇 번째 이후의 값을 참조할지 지정할 수 있습니다.

예시 쿼리

아래 쿼리는 직원의 고용일을 기준으로 다음 직원의 급여를 가져옵니다. 또한 *LAG() *함수와 비슷하게, 다음 행이 없을 경우 기본값으로 0을 반환합니다.

SELECT employee_id, salary,
       LEAD(salary, 1, 0) OVER (ORDER BY hire_date) AS next_salary
FROM employees;

쿼리 실행 결과

마지막 직원의 경우 다음 행이 없기 때문에 next_salary는 0을 반환하고, 다른 행들은 각각 다음 직원의 급여를 참조합니다.

추가로, 이해를 돕기 위해 로그 데이터에서 사용자 행동의 다음 단계를 분석해야 하는 상황을 가정한다면, LEAD() 함수를 적용한 아래와 같은 쿼리를 활용하여, 사용자가 현재 페이지를 방문한 후에 다음 페이지로 어디를 방문했는지 추적할 수 있을 것입니다.

SELECT user_id, page, timestamp,
       LEAD(page, 1) OVER (PARTITION BY user_id ORDER BY timestamp) AS next_page
FROM web_logs;

TIP💡

LAG() *와 *LEAD() 함수는 SCD Type 2와 같은 데이터 히스토리 관리 방식에서 데이터를 시간순으로 비교하거나 추적하는 할 때도 유용합니다. 이러한 함수들은 데이터의 변경 내역을 시간별로 분석하고, 각 데이터의 이전 상태나 다음 상태를 확인하는 데 활용될 수 있습니다.

참고로, 이전 올렸던 DBT 관련 포스팅에서 SCD Type에 대해 설명하였으니, 아래 포스팅을 확인해주시면 감사하겠습니다 :) Why DBT? (feat. A/B 테스트, 데이터 품질)

참고문헌

• https://livesql.oracle.com/ords/livesql/file/tutorial_GMLYIBY74FPBS888XO8F1R95I.html
• https://livesql.oracle.com/ords/livesql/file/content_CZUCT0MCOQZMJM7TI553HC8S9.html
• https://dev.mysql.com/doc/refman/8.4/en/subquery-materialization.html
• https://dev.mysql.com/doc/refman/8.0/en/optimizer-hints.html
• https://www.stratascratch.com/blog/the-ultimate-guide-to-sql-window-functions/
• https://towardsai.net/p/l/databricks-upsert-to-azure-sql-using-pyspark
• https://schatz37.tistory.com/46

ETL을 하는 이유는 결국 ELT를 하기 위함이며, 이때 데이터에 대한 품질 검증이 중요해집니다.

데이터가 점점 대용량으로 가게되면, 데이터의 품질 이슈가 발생합니다. 이는 인사이트를 뽑아낼 때 속도가 지연되거나, 잘못된 결론을 내리는 이슈로 이어지기도 하므로, 데이터의 신뢰성과 향후의 인력 및 리소스 소모를 막기 위해서는 필수적으로 해결되어야 하는 문제라고 생각합니다.

일단 ‘데이터 품질’에 대한 정의를 정리하고 가려고 합니다.

데이터 품질이란?

최근 읽고 있는 책인 조 라이스와 맷하우슬리의 ‘견고한 데이터 엔지니어링’에 따르면, 데이터 관리는 원천 시스템 단계에서 필수적이지만, 특히 변환 단계에서는 더 중요하다고 말합니다. 저자는 크게 3가지 이유에서 데이터 품질의 중요성을 강조합니다.

(참고로 해당 책은 데이터 엔지니어링 업무를 하고 계시거나, 준비 중이신 분들 이라면 꼭 추천드립니다! 현업에서 일부 경험했지만 굉장히 모호했던 개념들이 많았는데, 이를 모두 깔끔하게 정리해주는 좋은 책입니다 ^^)

정의적 정확성 (ex: semantic, metric)

일단, 변환 단계에서는 정의적 정확성을 고려하는 것에 대한 중요성을 강조합니다. 이는 변환이 예상되는 비즈니스 논리에 부합하는가? 에 대한 물음에서 파생된 체크포인트인데, 이를 지키기 위해 변환과는 독립적으로 존재하는 semantic, metric 계층이라는 개념이 점점 더 대중화되고 있습니다.

또한, 런타임 시 변환에서 비즈니스 로직을 적용하는 대신, 이러한 정의를 변화 계층 이전에 독립 실행형 단계로 유지하는 추세로 가고있다고 합니다. 이러한 문제를 효과적으로 해결할 수 있는 툴이 바로 Dbt입니다.

사실, 보통은 semantic, metric 계층을 구축하기 이전에 사용자에 대한 정보를 모으고, 값에 대한 명확한 표현 및 분리를 위해 fact, dimension 계층에 대한 테이블을 구성해놓습니다.

본 글에서는, 실제로 dbt를 통해 간단한 A/B 테스트 과정을 위한 파이프라인을 구축해 볼 예정인데, 이때 주로 사용할 계층인 fact 그리고 dimension 테이블에 대한 설명을 먼저 진행해보겠습니다. 이미 알고계신다면 넘어가셔도 좋습니다 :)

Fact 테이블

fact 테이블은 분석의 초점이 되는 양적 정보를 포함하는 중앙 테이블을 뜻합니다. 쉽게 말하면, fact 테이블은 값을 나타내는 테이블, dimension 테이블은 값을 설명하는 테이블이라고 생각하면 되고, fact 테이블은 일반적으로 dimension 테이블들과 외래 키로 연결되곤 합니다. (보통 fact 테이블의 크기가 훨씬 큽니다)

일반적으로 fact 테이블에는 비즈니스 결정에 사용될 매출 수익, 판매량, 이익과 같은 측정 항목을 포함시킵니다.

Dimension 테이블

dimension 테이블은 fact 테이블에 대한 상세정보, 즉 특정 개체의 속성값을 제공하는 테이블입니다. (예: 고객, 제품 과 같은 테이블) 따라서, fact 테이블의 데이터에 맥락을 제공하여 다양한 방식으로 분석가능하게 해줍니다. 위에서 언급했다시피, dimension 테이블은 보통 PK를 가지며, fact 테이블에서 참조합니다.

데이터의 결함

두번째로는, 변환에는 데이터의 변형이 수반되므로 사용 중인 데이터에 결함이 없고 실제 데이터를 정확히 나타내는지 확인하는 것이 매우 중요하다고 합니다. 저는 이를 일종의 ‘데이터 정합성’을 지켜야 한다는 것에 대한 강조라고 생각합니다. 본 글에서는 데이터 정합성에 대한 논의는 중점적으로 진행하지 않을 것이긴 한데, 많은 사람들이 데이터 정합성에 대한 정의를 다른 개념과 혼동하는 것 같아 한 번 정리해보려고 합니다.

데이터 정합성이란?

일반적으로 데이터 정합성이라고 하면, data set에 대한 결손값이나, 중복값, 혹은 이상값을 얼마나 잘 검증했는지를 뜻한다고 생각합니다. 엄밀히 개념을 설명하자면, 데이터 정합성이란 어떠한 데이터들에 대한 값이 서로 일치한지를 의미합니다.

데이터 정합성 검증 예시

보통, 원천 데이터에서 데이터를 가져올 경우 기본적으로 데이터가 올바르게 들어 왔는지에 대한 확인이 필요합니다. 이를 검증하기 위해서 OLTP → OLAP 형태의 데이터 파이프라인을 기준으로 설명하면 아래와 같은 2가지의 기본적인 검증 로직을 추가할 수 있습니다.

첫번째로, 원천 데이터와 목적지 데이터의 건수가 같은지 비교합니다.

원천 데이터에 로그 적재 시간이 적재되어 있다고 가정하면, created_date에 특정기간에 대한 조건을 걸어서, 레코드를 카운트하고 이를 비교하여 중복이나 결손값을 확인할 수 있습니다. 이때, 원천 데이터가 인덱스가 지원되는 OLTP 기반의 DB라면 created_date에 인덱스를 설정하면 부하를 줄일수 있습니다.

또한, 이 과정에서 전체 레코드수를 계산하지 않고 특정 기간에 필터링을 거는 이유는 보통 원천데이터 소스는 OLTP 기반의 DB에 저장되어 있기 때문에, 계속해서 데이터가 변하기 때문에 기간별로 필터링하는 것이라고 할 수 있습니다. 추가로, 이러한 모니터링 결과는 데이터의 증감률이나, 최근 데이터의 존재여부를 가시적으로 확인할 수 있다는 장점도 있습니다.

두번째로는, 목적지의 데이터의 유니크 키값을 이용하여 중복이 없는지를 비교합니다.

사실 앞서 검증한 데이터의 건수가 올바르다고 해도, 아래와 같은 SQL문을 통해 중복에 대한 검증은 추가로 진행되어야 합니다.

SELECT 
    id, 
    COUNT(*) as cnt
FROM 
    raw_data.user_event
GROUP BY id
HAVING COUNT(*) > 1

간단하게 데이터 정합성의 개념과 검증 방법에 대해 알아보았는데, 가장 중요한 것은 이러한 데이터를 검증 하기 위한 과정에서 비용이 많이 발생하거나 시스템의 부하를 줄 경우를 고려하며 개발을 진행해야 한다는 것입니다.

이제 본론으로 돌아와, 데이터 품질에 대한 3번째 논의로 넘어가보려고 합니다 :)

데이터 카탈로그

마지막으로, 데이터 변환 때문에 데이터 집합이 동일한 경로에서 어떻게 파생되었는지 알기 어려울 수 있다는 점이 있습니다. 이는 데이터 카탈로그 문제로 치환될 수 있는데, 이렇게 데이터의 계보를 유지하고 모니터링할 수 있는 것도 중요합니다. 이를 데이터 엔지니어가 직접 작업하고 운영한다는 것은 시간이 많이 소모되고 고된 일이라는 것을 아실 것입니다. 이러한 문제는 보통 오픈소스인 Datahub를 사용하여 해결합니다.

Dbt를 본격적으로 설명하기 전에, 먼저 짚고 넘어가야 할 히스토리를 유지하는 것에 대한 중요성과 5가지 SCD Type에 대한 개념을 정리해봅시다.

history를 왜 유지해야 할까?

OLAP 환경인 데이터 웨어하우스나 데이터 레이크에서 테이블들의 히스토리를 유지하는 것이 중요한 이유는 일부 속성들은 시간을 두고 변하게 되기 때문입니다.

보통은 created_at (생성시간으로 한번 만들어지면 고정)과 updated_at (마지막 수정 시간을 나타냄)과 같은 timestamp 필드를 생성하여 관리하는 것이 좋은데, 컬럼의 성격에 따라 이를 어떻게 유지할 지에 대한 방법이 또 달라집니다. 이를 설명하기 위한 개념인 SCD Type (Slowly Changing Dimension) 5가지에 대해 설명드리겠습니다.

SCD Type 0

SCD Type 0는 한번 쓰고 나면 바꿀 이유가 없는 경우들을 뜻합니다. 예를 들어, 유저의 회원등록일이나 제품 첫 구매일과 같이 첫 이벤트 발생 시에 정해지면 갱신되지 않고 고정되는 필드들이 있을 것입니다.

SCD Type 1

SCD Type 1는 데이터가 새로 생기면, 덮어쓰면 되는 컬럼들에 대한 특성입니다. 또한, 처음 레코드 생성시에는 존재하지 않았지만, 나중에 생기면서 채우는 경우도 이에 해당합니다.

예를들어, 연간 소득 필드의 경우 지속적으로 덮어쓰면 큰 이상이 없을 것이고, 고객이 초기에 이메일을 저장하지 않았을 때, 후에 업데이트하는 경우도 예시로 적합합니다.

SCD Type 2

SCD Type 2는 특정 entity에 대한 데이터가 새로운 레코드로 추가되어야 하는 경우입니다. 예를 들어, 이커머스 서비스를 사용중인 유저의 등급이 변화했다고 가정해봅시다.

이때, rank가 update된 사항은 중요한 데이터가 될 수 있으므로 변경시간을 같이 추가하여 데이터 품질을 유지할 수 있을 것입니다. 참고로 SCD Type 2 은 이후 dbt snapshot 기능을 사용하면서 한번 더 언급할 특성입니다 :)

SCD Type 3

SCD Type 3 는 SCD Type 2의 대안으로, 특정 entity 데이터가 새로운 컬럼으로 추가되는 경우를 뜻합니다.

위의 경우와 동일한 상황이라면, 아래와 같이 새로운 컬럼 (previous_rank)를 생성하여 데이터 품질을 유지할 수 있을 것입니다. 이 경우에도 변경시간 또한 별도 컬럼으로 존재해야 할 것입니다.

SCD Type 4

SCD Type 4 는 특정 entity에 대한 데이터를 새로운 Dimension 테이블에 저장하는 경우로, 일종의 SCD Type 2의 변종입니다. 예를 들어, 위 상황과 동일하다면 아래처럼 별도의 과거 이력 테이블을 생성하여, 아예 일반화하여 히스토리를 유지하는 방식입니다.

지금까지, 데이터 품질의 정의와 히스토리의 중요성, 그리고 SCD Type (Slowly Changing Dimension)에 대해 알아보았습니다. 특히, SCD Type에서 논의하는 지점들에 대해서 Dbt는 특정 세팅만 해주면 이와 같은 특성들에 대한 적절한 대처를 큰 스트레스(?)없이 효율적으로 진행할 수 있습니다.

그럼 이제 Dbt가 무엇인지, 어떤 방식으로 데이터 변환 과정에 관여하는지를 본격적으로 알아보겠습니다.

What is Dbt?

출처: https://docs.getdbt.com/docs/introduction

Dbt는 Data Build Tool의 약자로 ELT(Extract, Load, Transform)용 오픈소스 도구로, 데이터 웨어하우스 내에서 데이터 변환을 수행합니다. (ELT와 ETL의 개념에 대해서는 예전에 Data와 Data Engineer의 역할이라는 포스팅에서 짧게 언급하였으니 확인해주시면 감사하겠습니다 🙂)

여담이지만, Dbt의 등장으로 Analytics Engineer라는 직무 개념이 나왔다고 합니다.

우리가 흔히 알고 있는 BigQuery나 Snowflake와 같은 데이터 웨어하우스 솔루션들은 모두 지원하는 것으로 보입니다. 보통은 아래와 같이 Airlfow로 dbt를 스케줄링하고, DW와 연동된 dbt내의 여러 모델을 적절한 구조로 배치하고, 이를 실행한 결과를 통해 원하는 데이터를 확인할 수 있도록 데이터 파이프라인을 구성합니다.

dbt는 dbt Labs에서 제공하는 Cloud 버전도 존재합니다. 워낙, 툴 자체가 가볍기 때문에 우리가 늘 고려하는 DW 비용(?)처럼 큰 무리를 주지 않고, 직접 관리하는 리소스 비용을 생각하면 비용이 적절하여 많이 사용한다고 합니다. 본 글에서는, dbt를 직접 설치(dbt core)하여 로컬에서 작업해보려고 합니다.

출처: https://docs.getdbt.com/docs/cloud/about-cloud/architecture

Dbt의 특징

Dbt는 아래와 같은 특징들을 가지고 있어, 데이터 엔지니어의 여러 요구조건을 만족시킬 수 있습니다.

1. 데이터 변경 사항을 이해하기 쉽고 rollback 가능: SQL 기반의 변환 작업을 코드로 관리하여, 변경 사항을 쉽게 추적하고 rollback을 지원
2. 데이터 품질 테스트 및 에러 보고: 데이터 테스트 기능을 제공하여, 데이터 품질을 보장하고 에러를 사전에 감지 (ex: dbt test)
3. Fact 테이블의 Incremental Update: Incremental Update 기능을 지원하여, Fact 테이블의 데이터를 효율적으로 갱신 가능
4. Dimension 테이블 변경 추적 (history 테이블): SCD Type을 고려한 기능을 통해 Dimension 테이블의 변경 이력을 추적 가능

이외에도 dbt에서는, dbt docs generate 명령어를 통해 편리하게 documentation 기능을 활용할 수 있고, dbt docs serve 명령어를 통해 데이터간 리니지를 쉽게 확인할 수 있는 기능도 지원합니다만, 본 글에서는 위의 특징들을 중점적으로 활용하고, 실제 구현하는 방식에 대해 설명해보려고 합니다 😄

Dbt로 A/B 테스트해보기

A/B 테스트 분석을 쉽게 하기 위한 ELT 테이블을 만들어보자!

도입

먼저, Dbt 파이프라인을 구축하기 위해 DW는 AWS Redshift를 채택하였고, 입력테이블은 아래와 같이 생성하였습니다. (참고로, dummy_data는 미리 random한 데이터를 생성하여 csv형태로 테이블을 만든 뒤, 삽입하였습니다)

• user_event: 사용자, 날짜, 아이템 별로 impression이 있는 경우에 해당 정보를 기록하고, impression으로 부터 클릭, 구매, 구매 시 금액이 기록되는 데이터
  • 실제 환경에서는 이러한 aggregate정보를 로그 파일등의 소스로부터 만들어내는 프로세스가 필요할 것입니다.
• user_variant: 사용자가 소속한 AB test variant를 기록한 데이터입니다. (예: control vs test)
  • 보통은 experiment와 variant 테이블이 별도로 존재하고, 언제 variant_id로 소속되었는지를 기록하는 timestamp 필드가 존재하는 것이 일반적입니다.
• user_metadata: 성별, 나이 등의 메타정보를 담은 데이터
  • 이를 이용하여 여러 각도에서 AB테스트를 진행하여 다양한 인사이트를 도출할 수 있습니다.

최종적으로, 저희의 목표인 ELT 테이블 (생성테이블)은 미리 SELECT문으로 표현해보면, 아래와 같은 형태일 것입니다.

SELECT
    variant_id,
    ue.user_id,
    datestamp,
    age,
    gender,
COUNT(DISTINCT item_id) num_of_items, -- 총 impression
COUNT(DISTINCT CASE WHEN clicked THEN item_id END) num_of_clicks, -- 총 purchase
SUM(paidamount) revenue -- 총 revenue
FROM raw_data.user_event ue
JOIN raw_data.user_variant uv ON ue.user_id = uv.user_id 
JOIN raw_data.user_metadata um ON uv.user_id = um.user_id 
GROUP by 1, 2, 3, 4, 5;

생성 테이블: Variant 별 사용자에 대한 daily summary 테이블

• variant_id, user_id, datestamp, age, gender (5개의 필드에 대해 그룹바이)
• 총 impression, 총 click, 총 purchase, 총 revenue (sum 분석)

추가적으로, raw_data 스키마 이외의 분석을 위한 용도로 danie 라는 스키마를 생성하였습니다.

dbt 설치

이제, dbt를 직접 설치해야 하는데 저는 dbt에 대해 조금 더 깊게 알아보기 위해, dbt Cloud 대신 dbt Core를 직접 로컬에 설치하여 진행해보려고 합니다 :)

(늘 느끼는 것이지만, 공식문서보고 직접 설치 파일을 하나하나 뜯어보는 것이 나한테 가장 빠른 학습 방법인 것 같다ㅎ)

아래 명령어를 실행하여 dbt-redshift 를 설치하게 되면, dbt Core를 설치함과 동시에 redshift와의 연동을 쉽게 진행할 수 있습니다.

pip3 install dbt-redshift

참고로, python은 3.12x 버전, dbt는 1.8x를 사용하였고 각 버전 호환성은 아래 링크에 정리되어 있습니다.

What version of Python can I use? | dbt Developer Hub

dbt 프로젝트 생성

아래 명령어를 실행하게 되면, dbt 프로젝트를 생성함과 동시에 Redshift connection을 위한 config를 설정할 수 있습니다.

dbt init dbt_user_analysis

아래의 사진처럼, 프로젝트 디렉토리 dbt_user_analysis 는 dbt_project.yml, tests, snapshots, models 등을 포함하고 있습니다.

이제부터 dbt의 핵심 구성 및 기능이라고 할 수 있는 models, tests, snapshots 를 중심으로 글을 이어나가겠습니다.

dbt model

dbt model은 ELT 테이블을 만들 때 기본이 되는 빌딩 블록이고, Table이나 View 혹은 CTE의 형태로 존재합니다. 또한, model은 일종의 입력, 중간 그리고 최종 테이블을 정의하는 곳이라고 생각하면 되는데, 구체적으로 설명하면 아래와 같습니다.

dbt model은 raw, staging, core와 같은 일종의 계층에 대한 티어 개념이 존재하는데, raw → staging (src) → core의 순서로 이해하면 됩니다.

Input

입력(raw)과 중간(staging, src) 데이터 정의

• raw는 CTE로 정의
• staging은 View로 정의

Output

최종 (core) 데이터 정의

• core는 Table로 정의

최종적으로, 위와 같은 데이터 정의들은 models 디렉토리 아래에 SQL파일로 존재합니다.

이제 A/B 테스트를 위한 최종 ELT 테이블을 위해 raw에서부터 core까지 이어지는 과정을 총 3가지 단계를 통해 실제 구현해보겠습니다.

1단계

DW에 존재하는 raw_data를 기반으로 데이터 품질을 검증하여 Staging단계에 올리는 과정입니다. 데이터 용량이 매우 클 경우에는, incremental_update 타입으로 append시켜 유지하면 됩니다.

src (base) 테이블은 아래와 같이 정의하였습니다.

src_user_event.sql

WITH src_user_event AS (
    SELECT * FROM raw_data.user_event
) 
SELECT
    user_id, 
    datestamp, 
    item_id, 
    clicked, 
    purchased, 
    paidamount
FROM src_user_event

src_user_variant.sql

WITH src_user_variant AS (
    SELECT * FROM raw_data.user_variant
) 
SELECT
    user_id,
    variant_id 
FROM
    src_user_variant

src_user_metadata.sql

WITH src_user_metadata AS (
    SELECT * FROM raw_data.user_metadata
) 
SELECT
    user_id, 
    age, 
    gender, 
    updated_at
FROM 
    src_user_metadata 

초기 dbt 프로젝트를 생성하면 기본적으로 models 디렉토리에 example이 주어지는데, 이를 삭제하고 위의 src모델들을 추가하였고, dbt run을 실행해보면 아래와 같은 결과를 볼 수 있습니다.

dbt debug

dbt run

Redshift

2단계

위 사진과 같은 과정을 진행하기 전에, dbt에서 중요한 개념인 Materialization에 대해서 알아볼 필요가 있습니다.

What is Materialization?

Materialization 은 입력 데이터들을 연결해서 새로운 데이터를 생성하는 것 (방식)을 뜻하는데, 보통 여기서 추가 transformation이나 data cleanup을 수행합니다.

dbt는 4가지의 내장 materialization을 제공하는데, 이는 아래와 같습니다.

View

• 데이터를 자주 사용하지 않는 경우

Table

• 데이터를 반복해서 자주 사용하는 경우

Incremental (Table Appends)

• Fact 테이블
• 과거 레코드를 수정할 필요가 없는 경우
• Upsert 지원

Ephemeral (CTE)

• 한 SELECT에서 자주 사용되는 데이터를 모듈화하는데 사용

materialized format을 config를 통해서 테이블마다 정해줄 수도 있지만 dbt_project.yml 파일을 수정하여 진행해도 됩니다. 저는 dbt_project.yml의 models 부분을 아래와 같이 수정하여, 프로젝트의 테이블들은 기본적으로 view로 빌드되지만, dim 디렉토리에 있는 테이블들은 모두 table로 빌드되는 구조를 유지했습니다.

models:
  dbt_user_analysis:
    +materialized: view
    dim:
      +materialized: table

Jinja 템플릿

이제부터 model을 생성하는 SQL문에서는 Jinja 템플릿을 활용해 볼 것입니다.

dbt에서는 Jinja 템플릿을 기반으로 ref 태그와 config를 활용하여 dbt 작업의 효율성을 더할 수 있습니다.

아래의 SQL문에서는 ref 태그를 통해 dbt내의 다른 테이블들에 엑세스를 진행할 것이고, config문을 사용하여 materialized 종류와 입력을 할 때 스키마가 변경되었을 경우에 대응 전략을 정할 수 있는 on_schema_change 파라미터를 사용할 것입니다.

on_schema_change는 fail, sync_all_columns, ignore, append_new_columns 등이 있는데, 나머지 방식을 사용한다고 해도 성공보장이 없기 때문에, fail 처리가 가장 안정적인 방식일 것입니다.

fact_user_event.sql

그럼에도, 중복데이터가 생길 수 있기때문에 새로 생긴데이터만 incremental하게 업데이트 하려면 아래와 같이 별도의 where절을 사용하여 처리하면 된다.

{{ 
  config(
    materialized = 'incremental',
    on_schema_change='fail' 
  )
}}
WITH src_user_event AS (
    SELECT * FROM {{ ref("src_user_event") }}
) 
SELECT
    user_id, 
    datestamp, 
    item_id, 
    clicked, 
    purchased, 
    paidamount
FROM 
    src_user_event

WHERE datestamp is not NULL 
{% if is_incremental() %}
    AND datestamp > (SELECT max(datestamp) FROM {{ this }}) 
{% endif %}

dim_user_metadata.sql

WITH src_user_metadata AS (
    SELECT * FROM {{ ref('src_user_metadata') }}
) 
SELECT
    user_id, 
    age, 
    gender, 
    updated_at
FROM 
    src_user_metadata

dim_user_variant.sql

WITH src_user_variant AS (
    SELECT * FROM {{ ref('src_user_variant') }}
) 
SELECT
    user_id,
    variant_id 
FROM
    src_user_variant

추가로, config에서 incremental_strategy 파라미터도 설정할 수 있는데, 아래의 값들을 사용할 수 있습니다.

• append
• merge
• insert_overwrite

상황에 따라, unique_key와 merge_update_columns필드를 사용하기도 하므로, 이러한 사항들을 고려하여incremental_strategy 를 적절하게 사용하면 좋을 것 같다.

이제 dbt run 을 통해 실행해보면 아래와 같이 테이블이 생성된 것을 확인할 수 있다.

참고로 dbt compile은 SQL code 까지만 생성하고 실행하지는 않는다. 여기서 말하는 SQL code는 target디렉토리에 존재한다.

3단계

최종적으로 A/B 테스트의 결과를 도출하기 위한 작업을 진행해보겠습니다.

어느 variant에 속해있는지, 성별과 연령은 어떻게 분포하는지를 보기 위해 먼저, dim_user_variant와 dim_user_metadata를 JOIN하여 dim_user 테이블을 구성하려고 합니다.

dim_user.sql

WITH um AS (
    SELECT * FROM {{ ref("dim_user_metadata") }}
), uv AS (
    SELECT * FROM {{ ref("dim_user_variant") }}
) 
SELECT
    uv.user_id, 
    uv.variant_id, 
    um.age, 
    um.gender
FROM 
    uv
LEFT JOIN um ON uv.user_id = um.user_id

마지막으로, impressions, clicks, 구매 등에 대한 통계를 살펴보기 위해 dim_user와 fact_user_event를 조인하여 새로 생성한 analytics 디렉토리 밑에 analytics_variant_user_daily 테이블을 구성합니다.

analytics_variant_user_daily.sql

WITH u AS (
    SELECT * FROM {{ ref("dim_user") }}
), ue AS (
    SELECT * FROM {{ ref("fact_user_event") }}
) 
SELECT
    variant_id,
    ue.user_id,
    datestamp,
    age,
    gender,
COUNT(DISTINCT item_id) num_of_items, 
COUNT(DISTINCT CASE WHEN clicked THEN item_id END) num_of_clicks, 
SUM(purchased) num_of_purchases, 
SUM(paidamount) revenue 
FROM 
    ue 
LEFT JOIN u ON ue.user_id = u.user_id GROUP by 1, 2, 3, 4, 5

이제 dbt run 을 통해 실행해보면 A/B 테스트를 위한 최종적인 테이블이 생성된 것을 확인할 수 있습니다.

dbt snapshots

dimension 테이블은 성격에 따라 데이터 변경이 자주 생길 수 있으므로 히스토리를 유지하는 것이 중요합니다.

dbt에서 snapshot은 테이블의 변화를 계속적으로 기록함으로써 과거 어느 시점이건 다시 돌아가서 테이블의 내용을 볼 수 있는 기능을 이야기 합니다. 이를 통해, 테이블에 문제가 있을경우 과거데이터로 rollback이 가능하고, 다양한 데이터 관련 문제에 대한 효율적인 디버깅 과정이 가능하게 합니다.

스냅샷을 사용하면 글의 서두에서 언급했던 SCD Type 2 와 같은 특성에 대해 히스토리를 유지하며 데이터 품질을 보장할 수 있습니다.

기존 Dimension 테이블에서 특정 entity에 대한 데이터가 변경되는 경우 새로운 Dimension 테이블을 생성하여 히스토리를 유지하는데, 구체적인 과정은 아래와 같습니다.

기본 구조는 PK를 기준으로 변경시간이 현재 DW에 있는 시간보다 미래인 경우를 변경 감지 기준으로 삼고, updated_at을 기준으로 새로운 데이터가 업데이트되면 히스토리 테이블에 append하게 됩니다.

snapshots 디렉토리에 아래와 같은 scd_user_metadata.sql을 작성하고, dbt snapshot 명령어를 실행하면 히스토리 테이블이 생성된 것을 확인할 수 있습니다.

scd_user_metadata.sql

{% snapshot scd_user_metadata %}

{{ 
  config(
    target_schema='danie',
    unique_key='user_id', 
    strategy='timestamp', 
    updated_at='updated_at', 
    invalidate_hard_deletes=True
  ) 
}}
SELECT * FROM raw_data.user_metadata

{% endsnapshot %}

dbt snapshot

Redshift

dbt tests

여기서 말하는 테스트는 일반적으로 소프트웨어에서 말하는 테스트가 아닌 일종의 데이터 품질을 테스트하는 방법을 뜻합니다.

dbt test의 종류를 나누자면, 아래와 같이 2가지로 나눌 수 있다.

Generics Test: 내장 테스트

Generics test는 Airflow operator처럼 꺼내쓸 수 있는 일종의 dbt 내장 테스트인데, unique, not_null, accepted_values, relationships 등의 테스트를 지원합니다.

models 디렉토리에 yaml 형태로 테스트 파일을 생성하면 되는데, 저는 아래와 같이 구성하였습니다.

schema.yml

version: 2
models:
- name: dim_user_metadata
columns:
- name: user_id
tests:
- unique - not_null

이를 테스트로 활용하기 위해서는 dbt test 명령어를 실행하여 아래와 같은 결과를 확인할 수 있습니다.

Singular Test: 커스텀 테스트

기본적으로 SELECT로 간단하게 실행할 수 있고, 결과가 리턴되면 “실패”로 간주되는 테스트입니다.

tests 디렉토리에 생성하고, 있으면 안될 것들이 있는지 확인하는 정도로 활용합니다.

예를들어, Primary Key Uniqueness 테스트 (물론, generic test로 쉽게 검증 가능하지만 ^^)를 진행한다고 하면, tests 디렉토리에 아래와 같이 dim_user_metadata.sql파일을 작성하면 됩니다.

dim_user_metadata.sql

SELECT *
FROM ( 
    SELECT
        user_id, 
        COUNT(1) cnt 
    FROM
        danie.dim_user_metadata
    GROUP BY 1
    ORDER BY 2 DESC
    LIMIT 1
)
WHERE cnt > 1

이제 위의 generic test에서 진행한 것과 동일하게 dbt test 명령어를 수행하면 되지만, 이렇게 되면 방금 진행했던 test들도 포함되어 실행되니, 아래처럼 특정 테이블을 지정하면 관련 테이블들에 대한 검증만 진행할 수 있습니다.

dbt test --select dim_user_metadata

본 글을 작성하며 개발한 사항은 아래 github public repo에 업로드하였으니 참고바랍니다 :)

참고문헌

• 견고한 데이터 엔지니어링 (written by Joe Reis & Matt Housley)
• dbt Developer Hub
• RDBMS 데이터 적재 시 데이터 정합성 체크

MySQL, MongoDB, BigQuery 등 많은 종류의 DB를 사용해보고 수많은 쿼리를 작성해보았지만, 정확히 아킥텍쳐와 실행원리를 뜯어본 적은 없는 것 같아 각각의 DB (MySQL, MongoDB, BigQuery에 대하여 알아보고 비교해보려고 합니다. 본 글은 RealMySQL 8.0, 그리고 말미에 언급되는 레퍼런스를 참고하여 작성되었습니다 :)

MySQL

MySQL은 가장 대표적인 RDBMS (관계형 데이터베이스) 시스템이라고 할 수 있다. 먼저 MySQL 공식문서에서 제공하는 대략적인 아키텍쳐는 아래 그림과 같다.

크게 MySQL 엔진(두뇌 역할)과 스토리지 엔진 (팔, 다리 역할)으로 분류할 수 있다. MySQL 엔진에는 클라이언트의 요청을 처리하는 커넥션 핸들러와 쿼리파서, 전처리기, 옵티마이저 등이 있고, 스토리지 엔진에는 InnoDB (default) 엔진과 MyISAM 엔진이 있다.

쿼리 실행 구조

일단 우리가 파이썬에서 커넥션을 위한 라이브러리를 사용하든, Shell에 바로 접근하든 MySQL서버에서 쿼리를 실행할때의 과정은 아래와 같을 것이다.

1. 쿼리파서: 쿼리 파서는 클라이언트 요청으로 들어온 쿼리를 토큰으로 분리하여 트리 형태의 구조로 만들어낸다. 기본적인 문접오류는 이 과정에서 발견된다.
2. 전처리기: 위 과정에서 만들어진 파서 트리를 기반으로 쿼리 문장에 구조적인 문제점이 있는지 확인한다. 실제 존재하지 않거나 권한 상 사용할 수 없는 개체의 토큰은 해당 단계에서 걸러진다.
3. 옵티마이져: 요청으로 들어온 쿼리 문장을 저렴한 비용으로 가장 빠르게 처리할지를 결정한다.
4. 실행 엔진: 옵티마이저가 회사의 경영진이라면, 실행엔진은 중간관리자, 핸들러는 각 업무의 실무자로 비유할 수 있다.
5. 핸들러 (스토리지 엔진): 핸들러는 MySQL 실행 엔진의 요청에 따라 데이터를 디스크를 저장하고 디스크로부터 읽어 오는 역할을 담당한다.

아래는 책에서 소개한 옵티마이저가 GROUP BY를 처리할 때, 임시테이블을 사용했다고 가정한 예시입니다.

1. 실행 엔진이 핸들러에게 임시테이블을 만들라고 요청
2. 다시 실행 엔진은 WHERE 절에 일치하는 레코드를 읽어오라고 핸들러에게 요청
3. 읽어온 레코드들은 1번에서 준비한 임시 테이블로 저장하라고 다시 핸들러에게 요청
4. 데이터가 준비된 임시 테이블에서 필요한 방식으로 (예: GROUP BY) 데이터를 읽어 오라고 핸들러에게 다시 요청
5. 최종적으로 실행 엔진은 결과를 사용자나 다른 모듈로 넘김

MySQL 스레딩 구조

MySQL 서버는 스레드 기반으로 작동하며, 크게 포그라운드 스레드와 백그라운드 스레드로 구분된다. 실행 중인 스레드의 목록은 performance_schema 데이터베이스의 threads 테이블에서 확인할 수 있다.

포그라운드 스레드

포그라운드 스레드는 최소한 MySQL 접속된 클라이언트의 수만큼 존재하며, 주로 각 클라이언트 사용자가 요청하는 쿼리 문장을 처리한다. 그래서 포그라운드 스레드를 클라이언트 스레드라고도 부른다.

클라이언트 사용자가 작업을 마치고 커넥션을 종료하면 해당 커넥션을 담당하던 스레드는 스레드 캐시로 되돌아간다. 이미 스레드 캐시에 일정 개수 이상의 대기 중인 스레드가 있으면 스레드 캐시에 넣지 않고 스레드를 종료시켜 일정 개수의 스레드만 스레드 캐시에 존재하게 한다. 이때 스레드 캐시에 유지할 수 있는 최대 스레드 개수는 thread_cache_size 시스템 변수로 설정한다.

포그라운드 스레드는 데이터를 MySQL의 데이터 버퍼나 캐시로부터 가져오며, 버퍼나 캐시에 없는 경우에는 직접 디스크의 데이터나 인덱스 파일로부터 데이터를 읽어와서 작업을 처리한다. MyISAM 테이블은 디스크 쓰기 작업까지 포그라운드 스레드가 처리하지만, InnoDB 테이블은 데이터 버퍼나 캐시까지만 포그라운드 스레드가 처리하고, 나머지 버퍼로부터 디스크까지 기록하는 작업은 백그라운드 스레드가 처리한다.

백그라운드 스레드

앞서 언급한 것과 같이 백그라운드 스레드는 MyISAM과는 연관이 없는 사항이지만, InnoDB는 다음과 같이 여러 가지 작업이 백그라운드로 처리된다.

• 인서트 버퍼를 병합하는 스레드

• 로그를 디스크로 기록하는 스레드
• InnoDB 버퍼 풀의 데이터를 디스크에 기록하는 스레드
• 데이터를 버퍼로 읽어오는 스레드
• 잠금이나 데드락을 모니터링하는 스레드

위의 작업 중 가장 중요한 것은 로그 스레드와 버퍼의 데이터를 디스크로 내렸는 작업을 처리하는 쓰기 스레드이다. MySQL 5.5 버전부터 데이터 쓰기 및 읽기 스레드의 개수를 2개 이상 지정할 수 있게하여, innodb_write_io_threads와 innodb_read_io_threads 시스템 변수로 스레드의 개수를 설정할 수 있다. InnoDB에서도 데이터를 읽는 작업은 주로 포그라운드 스레드에서 처리되기 떄문에 읽기 스레드는 많이 설정할 필요가 없지만 쓰기 스레드는 아주 많은 작업을 처리하기 때문에 일반적인 내장 디스크를 사용할 때는 2~4 정도, DAS나 SAN과 같은 스토리지를 사용할 때는 디스크를 최적으로 사용할 수 있을 만큼 풍분히 설정하는 것이 좋다고 한다.

사용자의 요청을 처리하는 도중 쓰기 작업은 버퍼링되어 처리될 수 있지만, 읽기 작업은 절대 지연될 수 없다. 책에서는 사용자가 SELECT 쿼리를 실행했는데, 10분 뒤에 결과를 돌려주겠다하는 데이터베이스는 없다는 예시를 들어 설명한다.

그래서 일반적인 DBMS에는 대부분 쓰기 작업을 버퍼링해서 일괄 처리하는 기능이 탑재돼 있으며, InnoDB 또한 이러한 방식으로 처리한다. InnoDB에서는 INSERT, UPDATE, DELETE 쿼리로 데이터가 변경되는 경우 데이터가 디스크의 데이터로 완전히 저장될 때까지 기다리지 않아도 된다. 그러나, 앞서 포그라운드 스레드에서 설명했다시피 MyISAM 엔진은 포그라운드 스레드가 쓰기 작업까지 함께 하고, 일반적인 쿼리는 쓰기 버퍼링 기능을 사용할 수 없다.

MongoDB

MongoDB는 대표적인 NoSQL중 하나이다. 나또한 토론 채팅 서비스를 구축하면서, 또는 업무에서 빠르게 비정형 데이터 위주로 구성된 DB 구축하고 싶을 떄 사용하였던 데이터베이스이다. 아무래도 objectid를 사용하여 어플리케이션 서버에서 1대1로 매칭할 수 있으므로 빠른 개발이 가능한 것 같다.

일단, MongoDB의 기본적인 구성은 아래와 같습니다.

• Config 서버: 중개자 계층, 샤딩을 위한 메타 데이터를 저장한다. (데이터들의 위치 정보를 저장)
• Mongos 서버: MongoDB의 중개자 역할, Config 서버의 메타 데이터를 이용해 각 MongoDB에 데이터 접근을 도와준다.(라우터와 같은 역할)
• Mongod 서버: MongoDB의 데이터 서버로써, 서버 장애에 대비해 MongoDB 서버 안에 여러 개의 리플리카 셋 구조로 구성되어 있다.

What is DocumentDB?

일단 MongoDB는 도큐먼트 데이터베이스라고 보통 불린다. 표면적으로는, JSON 형식으로 데이터를 관리하고, 도큐먼트 단위로 데이터를 저장하기 때문일 것이다. 보통, 도큐먼트는 관계를 가지는 데이터를 중첩 도큐먼트와 배열을 사용하여 1개의 도큐먼트로 표현한다.

{
  "name": "John Doe",
  "age": 30,
  "isStudent": false,
  "courses": ["Math", "Science", "History"],
  "address": {
    "street": "123 Main St",
    "city": "Anytown",
    "zipcode": "12345"
  }
}

BSON 형식

JSON은 구문 분석 속도, 타입 명확성 부족, 공간효율성 등의 면에서 단점을 가지고 있다. 따라서, 위와 같은 단점을 보완하기 위해 MongoDB는 BSON (Binary JSON) 형식을 도입하였다고 한다. 그 결과, MongoDB에서 우리가 눈으로 데이터를 확인할 때는 JSON으로 보이지만, 나머지 상황에서는 모두 BSON 형태로 저장하고 전송한다고 한다. (MongoDB 초기에는 모두 JSON으로 관리하였다고 한다)

BSON의 장점

BSON(Binary JSON)은 JSON의 단점을 보완하기 위해 고안된 바이너리 형식의 데이터 포맷이다. BSON은 다음과 같은 장점을 가지고 있다.

1. 빠른 구문 분석 속도

• BSON은 이진 포맷이기 때문에, 컴퓨터가 데이터를 읽고 해석하는 데 더 효율적입니다. 이진 데이터를 직접 읽고 필요한 위치로 이동할 수 있기 때문에 구문 분석 속도가 빠르다.
• BSON은 데이터를 타입과 함께 저장하므로, 파싱 시 데이터 타입을 명확히 알 수 있어 추가적인 변환 과정이 필요 없다.

2. 공간 효율성

• BSON은 데이터 타입 정보를 포함하여 저장하므로, 숫자, 날짜, 바이너리 데이터 등을 효율적으로 저장할 수 있다.
• 키 이름을 길게 반복하는 대신, BSON은 짧은 형식으로 데이터를 저장하여 공간을 절약할 수 있다.
• BSON은 정수, 부동 소수점 등의 숫자 데이터를 효율적인 이진 형식으로 저장한다. 예를 들어, 정수 1234는 4바이트로 저장되며, 이는 텍스트 형식보다 공간 효율적이다.

아래는 카카오 개발 컨퍼런스에서 MongoDB에 대하여 발표한 내용을 중심으로 정리한 글입니다. MongoDB 개념을 이해하고, 실제 카카오에서는 어떤 방식으로 MongoDB를 사용하는지 소개하는 유익한 영상이므로 MongoDB 입문자라면 한번씩 보시는걸 추천드립니다 :) https://tv.kakao.com/channel/3693125/cliplink/414072595

MongoDB의 특징

MongoDB의 4가지 특징 (신뢰성, 확장성, 유연성, 인덱싱 지원)을 가지고 있습니다.

Reliability: 서버 장애에도 서비스는 계속 동작

일반적으로 MongoDB는 1개의 primary와 2개의 secondary로 구성된 레프리카셋 구조를 가지고 있어 데이터 복제와 고가용성을 구현하기 때문에 장애로부터 안정된 상태를 유지한다. (primary, secondary는 master나 slave라는 용어로도 쓰이기도 한다)

만약 primary에서 장애가 발생하면 secondary가 primary가 된다. 따라서, 어느 한 서버에서 장애가 발생해도 데이터 유실을 막을 수 있고 application 서버는 별도로 이에 대한 처리를 하지 않아도 된다!

Scalability: 데이터와 트래픽 증가에 따라 수평확장 가능

MongoDB는 데이터 증가로 더이상 하나의 레플리카셋에 못담을 상황일 때, 데이터를 샤딩하여 분산 시켜준다. 또한, 이러한 과정이 서비스 중단없이 온라인으로 진행된다. 사실 MongoDB에서 auto-sharding을 지원한다고 했을 때 구체적으로 어떤 것을 의미하는지 몰랐는데, 이러한 점을 뜻하는 것으로 보인다)

위와 같은 과정을 용어로는 밸런싱 기능이라고 하는데, 이는 특정샤드에 데이터가 몰리면 다른 샤드로 데이터를 옮겨 전반적으로 모든 샤드가 균등하게 데이터를 저장할수 있게 하는 것을 뜻한다.

또한, 온라인상에서 데이터를 밸런싱하기 때문에 단일 레플리카셋에서 샤드로의 온라인 전환이 가능하며, 샤드의 확장 축소 모두 온라인에서 진행할 수 있다.

위에서 MongoDB의 구성에 대하여 언급하였는데, 구체적으로 어떤 과정으로 샤드클러스터에서 데이터를 다루는지 살펴보자.

0. 샤드 클러스터에 저장되는 실제 데이터는 각 샤드 1, 2, 3에 나누어 저장

1. 어떤 데이터가 어떤 샤드에 있는지는 config 서버에 저장
2. application 서버는 mongos 서버를 통해 샤드 클러스터에 접근
3. mongos 서버는 config 서버와 통신하여 요청받은 데이터가 어느 샤드에 있는지 확인하고 해당 샤드에서 데이터를 조회하여 application에 전달

application단에서는 샤드 클러스터 내부 동작을 알 필요없이 위와 같은 일련의 과정을 mongos 서버에서 알아서 해주기 때문에, application에서의 접근을 쉽게 만들어 준다.

Flexibility: 여러가지 형태의 데이터를 손쉽게 저장

MySQL과 같은 RDBMS의 경우, 새로운 특성을 추가하려면 컬럼을 별도로 추가해야 한다. 그러나, MongoDB는 스키마를 제공하지 않으므로 데이터 변경에도 유용하게 대처 가능하다. 예를 들어, 고객의 핸드폰 번호를 담는 테이블이 있을 때 기존에 없던 기기 OS 정보를 추가하고 싶을 때 혹은 핸드폰이 여러개일 때, 테이블을 따로 추가하지 않고 배열로 그냥 담아버리면 된다.

참고로 RDMBS에서의 데이터 단위와 MongoDB에서의 데이터 단위에 대한 대응관계는 아래와 같다.

이와 같은 특성으로, MongoDB는 데이터 구조를 한눈에 볼수 있고 application에서 다루는 객체와 1대1대응 관계로 이루어져 있어 개발자는 쉽게 데이터를 이해하고 빠르게 개발할 수 있다.

Index Support: 다양한 조건으로 빠른 데이터 검색

보통의 NoSQL에서는 데이터를 찾고 분산할 목적으로 키(PK)를 한개만 제공한다.

nosql에 customer_id를 PK로 지정한다고 가정하면, 데이터는 여러 서버에 customer_id를 기준으로 나누어서 저장할 것이다. 때문에, 특정 id값으로 검색하면 데이터가 저장된 서버를 알 수 있고 바로 조회가 가능하다. 그러나, 이름을 기준으로 검색할 상황이 있을 때는 해당 데이터 (정확히는 documents)가 어느 서버에 위치하는지는 모르기 때문에 모든 서버를 검색해야 하고, 대용량 데이터일 경우 비용은 매우 클 것이다.

MongoDB의 다양한 인덱스 제공 기능은 위와 같은 이슈를 방지할 수 있다. 필요한 필드에 필요한 만큼 인덱스를 생성할 수 있으므로, 위 예시의 경우에는 이름 field에도 인덱스를 생성하여 데이터를 빠르게 찾을 수 있다.

MongoDB가 제공하는 다양한 형태의 인덱스는 아래와 같다.

• Hashed 인덱스: 샤드 클러스터에서 데이터를 균등하게 분산하고자 할 때 사용하는 인덱스
• TTL 인덱스: 제한시간을 설정하여 오래됀 데이터를 자동으로 지워주는 인덱스; 보관 기간이 정해진 데이터는 어플리케이션에서 굳이 관리 하지 않아도 된다.
• Geospatial 인덱스: 공간 내의 거리나 범위를 다루기 위해 사용하는 일종의 공간 인덱스; 카카오 모빌리티 서비스에서 사용한다.
• Multikey 인덱스, Partial 인덱스...etc

사용사례

해당 발표에서 MongoDB의 여러 사용사례를 보여주었는데, 대용량 로그 저장 및 조회를 위해MongoDB를 도입한 사례와 MySQL에서 MongoDB로 이전한 사례가 MongoDB의 이해에 매우 도움될 것 같아 본 글에서 소개드리려고 합니다 :)

대용량 로그 저장 및 조회를 위한 MongoDB를 도입

보통, 사용자의 요청은 로그로 저장되며 통계를 분석하여 서비스 개선에 활용하는데, 통계를 계산할 시에 특정 기간에 대한 전체 데이터를 읽는 용도로 HBase가 많이 사용됩니다.

하지만, 통계가 아니라 이름이나 물품명으로 등으로 검색해야 할 시에, HBase는 PK만 지원하므로 키가 걸리지 않은 데이터라면 전체 데이터를 읽어야 하고, 이는 높은 비용을 발생시킬 가능성이 농후합니다.

위와 같은 상황에서, MongoDB를 도입하여 필요한 필드에 필요한 만큼 인덱스를 부여할 수 있기 때문에 대용량 데이터가 있어도 빠르게 데이터를 찾을 수 있습니다.

따라서, application에서 로그를 저장할 때 MongoDB에 저장하거나, HBase의 로그데이터를 주기적으로 MongoDB에 업데이트하면 관리자가 다양한 조건으로 원하는 데이터를 빠르게 조회할 수 있습니다.

DB 이전 사례 (MySQL → MongoDB)

사내 서비스 중 서비스가 확장되면서 기존의 MySQL로 유지하던 서비스의 한계점이 아래와 같이 나타났다고 합니다.

• 상품데이터와 로그데이터의 혼재
• 수TB의 디스크 -> Scale-Up 한계
• 테이블 당 수백 GB -> 스키마 변경시마다 10시간 이상 소요

아무래도 RDBMS 특성 상 마지막 한계점은 극복하기 어려우므로 NoSQL로의 이전은 불가피 했을 것으로 보인다. 따라서 MongoDB로 이전하였고, 아래와 같은 성과를 보였다고 합니다.

• 로그데이터 이관 -> 63% 압축률
• 스키마 변경 부담 제거
• 샤드 클러스터 구성 -> 서비스의 확장에 따라 유연하게 Scale-Up

아마 압축 방식은 default로 방식을 사용하였다고 했는데, 아마도 snappy 방식 일 것이다. 아래는 MongoDB에서 지원하는 여러 압축 방식인데, MongoDB는 snappy 방식이 기본값으로 설정되어있다.

BigQuery

BigQuery는 RedShift, Snowflake와 같은 데이터웨어하우스 솔루션 중 하나로, 그 중 현재 가장 많이 쓰이는 시스템이라고 볼 수 있다. 데이터웨어하우스는 대용량 데이터를 처리하기 위한 빅데이터 기반 데이터베이스 이므로, 굳이 데이터의 양이 크지 않다면 사용하지 않아도 된다. (오히여, 성능 저하가 올 수 도 있음)

BigQuery의 표면적인 특징은 아래와 같습니다.

• SQL로 데이터 처리 가능 (Nested fields, Repeated fields)
• CSV, JSON, Avro, Parquet 등과 같은 다양한 데이터 포맷을 지원
• 구글 클라우드 내의 다른 서비스들과 연동이 쉬움
• 배치 데이터 중심이지만 실시간 데이터 처리 지원
• 스키마 변경도 유연하게 대처 가능

사실 대부분의 클라우드 기반의 데이터웨어하우스 솔루션들은 위 특징들을 만족한다. 특히 3번째 특징인 동일한 클라우드 내에서 다른 서비스들과 연동들이 쉬운 부분이 개발 생산성에 큰 기여를 한다. 예를 들어, AWS내에서 AWS Athena + S3 + RedShift 조합이나, GCP내에서 Cloud Storage + BigQuery + Cloud Scheduler 조합을 사용하여 파이프라인을 구축한다면 리소스 소모가 매우 줄어들 수 있다. 물론, 필요성과 성능 및 비용을 고려하여 파이프라인 구성을 제작하는 것이 선결과제 임을 잊으면 안된다 :)

다음 step으로 넘어가기 전에, BigQuery를 사용할 때 알고 있으면 좋은 2가지를 설명해보려고 합니다.

• PK 제공 X: 보통의 데이터웨어하우스 솔루션들은 동일한 사항인데, RDBMS와 같이 인덱스나 PK와 같은 키는 제공되지 않는다. 애초에 대용량 처리를 위한 데이터웨어하우스 설계 목적에 반하기 때문이라고 생각된다. 따라서, 보통 Full scan으로 진행됩니다.

• Delete 불가: BigQuery는 Delete문을 지원하지 않고, append 하는 것만 지원한다. 따라서, 한번 입력된 데이터는 변경되거나 삭제될 수 없습니다. 나 또한 실제로 BigQuery를 사용하며 필드를 잘못 추가하여 데이터가 몇개 유입되는 바람에 테이블을 지우고 다시 생성한 적이 있다ㅠ 근데, BigQuery는 Update문도 지원하지 않는다고 들었는데 나는 얼핏 단발성으로 사용했던 기억이 있는 것 같은데...? 추후에 관련 경험을 공유해보겠다 :)

그럼 왜 기존의 데이터베이스를 쓰지 않고, 데이터 웨어하우스를 꼭 사용해야 하는걸까? 기존 데이터베이스는 읽기/쓰기 작업과 분석 작업에서 리소스 충돌이 발생할 수 있지만, BigQuery는 컴퓨팅 레이어와 스토리지 레이어를 분리하여 각 레이어가 독립적으로 성능과 가용성을 유지하며 동적으로 리소스를 할당할 수 있기 때문이다. 또한, BigQuery 스토리지는 고가용성을 위해 여러 위치 간에 자동으로 복제됩니다. 물론, 이 부분은 MongoDB도 유사한 특징을 가지고 있다.

그렇다면, 이제부터 구체적으로 왜 BigQuery가 대용량 데이터에 적합한 솔루션인지 3가지 특징을 통해 알아보자. 아마, 클라우드 기반의 데이터웨어하우스 특성에 대한 전체적인 이야기가 될 수 있다.

열 기반 스토리지 (Columnar Storage)

이해하기 쉽게 RDBMS를 사용하는 사례를 하나 가정해보겠습니다. 만약, MySQL에서 아래의 쿼리를 실행한다고 생각해보자.

select product_id, client_id 
from payment_table 
where ~~~~

아무리 특정 컬럼을 select하고 where에 조건을 걸었다 하더라도, 일단 SSD에서 테이블 전체를 읽어와서 메모리에 올린 다음에 attribute를 필터링한다는 것을 알고 있을 것이다. 이때 발생하는 I/O 부담을 줄이려고, 우리는 보통 partitioning을 사용한다. 어쨋든, 이러한 이슈가 발생하는 이유는 일반적으로 RDBMS가 Row 기반으로 설계되었기 때문일 것이다. (필요한 컬럼만 똑딱떼어올 수 없다는 뜻이다)

하지만, BigQuery는 레코드 별로 저장하는 것이 아니라 컬럼 별로 저장하는 열 기반 스토리지 형식을 갖추고 있기 때문에 위와 같은 상황에서는 유리한 위치를 점한다.

아래의 그래프는 관련 논문에서 single-field에 대한 접근을 기준으로 실험한 결과이다. 참고로, Dremel은 Google의 BigQuery 서비스 에서 사용되는 쿼리 엔진이다.

추가로, 컬럼 별로 같은 데이터 타입들로 이루어진 데이터들이 모아져있어, 컬럼별 압축률도 우수하며 컬럼을 추가하거나 삭제하는 것도 매우 빠릅니다. MongoDB 설명 말미에 각 압축 방식에 따른 Compression ratio에 대한 그림을 확인해보면 가장 높은 압출률이 10을 넘지를 못하지만, BigQuery의 경우 상회할 수 있다고 한다.

스키마 구조 (feat. Repetition Level, Definition Level)

서두에 BigQuery의 표면적인 특징을 나열하면서, 스키마 변경도 유연하게 대처가 가능하다고 언급하였다. 아마 이런 의문이 들 수 있다. "중첩된 데이터에 대해 MongoDB는 Document-based DB니까 대처가 쉽게 가능할 것 같긴 한데, 열 지향 기반으로 설계된 BigQuery가 어떻게...?"

BigQuery는 중첩된 필드(Nested fields)와 반복된 필드(Repeated fields)를 효율적으로 저장하고 쿼리하기 위해 Repetition Level과 Definition Level을 사용한다. 이 두 개념을 통해 중첩된 데이터 구조를 열 지향 방식으로 평탄화하여 저장하고, 이를 효율적으로 쿼리할 수 있다.

일단 BigQuery가 이러한 구조를 어떻게 처리하는지 이해하기 위해, 먼저 논문에서 제공하는 두 개의 샘플 레코드(r1과 r2)와 Nested fields와 Repeated fields의 개념을 정리해보자.

What is Nested fields & Repeated fields?

1. Nested fields: 중첩된 필드는 한 레코드 내에 또 다른 레코드를 포함하는 구조를 말한다. 예를 들어, Name 필드 안에 Language 필드가 중첩된 구조를 가질 수 있다.

2. Repeated fields: 반복된 필드는 한 필드 내에 여러 값을 가질 수 있는 구조를 말한다. 예를 들어, Links 필드 안에 여러 개의 Forward 링크를 포함할 수 있다.

Repetition Level과 Definition Level

먼저, Repetition Level과 Definition Level의 개념은 아래와 같다.

1. Repetition Level (r level): Repetition Level은 반복된 필드가 몇 번째 반복인지를 나타낸다. 중첩된 반복 구조에서 각 레벨의 반복 횟수를 표현한다.
2. Definition Level (d level): Definition Level은 특정 값이 정의되었는지 여부를 나타낸다. 중첩된 구조에서 각 필드가 정의된 깊이를 표현한다.

이제, 아래의 사진에서 Repetition Level과 Definition Level의 역할을 정리해보자.

Repetition Level는 중첩된 필드나 반복된 필드가 여러 번 반복될 때 각 반복의 순서를 나타낸다. 예를 들어, Links.Forward 필드가 세 번 반복될 때 각 값의 반복 수준은 0, 1, 2가 된다.

Definition Level는 필드가 정의되었는지 여부와 정의된 깊이를 나타낸다. 예를 들어, Name.Language.Country 필드가 정의되지 않은 경우, 해당 값은 NULL로 표시되며 Definition Level은 정의되지 않은 깊이를 나타낸다.

이와 같은 변환 과정을 통해, BigQuery는 중첩된 데이터를 효율적으로 쿼리할 수 있게 된다. 그 결과, BigQuery는 열 지향 기반 데이터베이스임에도 불구하고 복잡한 계층적 데이터를 효과적으로 처리할 수 있다.

트리 기반 분산 처리 (Tree Architecture Distribution)

Dremel 논문 기준으로 설명하면 전체적인 과정은 아래와 같을 것입니다. (root 서버는 mixer 0, intermediate 서버는 mixer 1이라는 용어로 쓰이기도 하니, 같은 개념이라고 이해해도 무관합니다)

1. 쿼리 입력: root 서버에 SQL 쿼리를 입력
2. 쿼리 분할: root 서버는 입력된 SQL 쿼리를 더 작은 SQL 문으로 분할하여 intermediate 서버로 전달
3. 중간 서버 처리: intermediate 서버는 root 서버로부터 받은 쿼리를 다시 더 작은 단위로 쪼개어 leaf 서버로 전달
4. leaf 서버 처리: leaf 서버는 실제 데이터를 저장하고 있는 파일 시스템에서 데이터를 읽어와 쿼리 연산을 수행
5. 결과 집계: leaf 서버는 연산 결과를 부모 노드(intermediate 서버)로 전달 -> intermediate 서버는 받은 결과를 집계하여 루트 서버로 전달
6. 최종 결과 반환: root 서버는 모든 결과를 취합하여 최종 쿼리 결과를 반환

실제 SQL 쿼리를 예시를 통해, 위 과정을 적용해보자.

SELECT
  customer_id,
  SUM(purchase_amount) AS total_purchase
FROM
  ecommerce_data_transactions
WHERE
  purchase_date BETWEEN '2020-01-01' AND '2022-12-31'
GROUP BY
  customer_id
ORDER BY
  total_purchase DESC
LIMIT
  5

1. 디스크에서 customer_id, purchase_date, purchase_amount 컬럼만을 읽어들입니다.
2. leaf 서버: 각 leaf 노드에서 읽어들인 데이터를 가지고 2020~2022년 기간의 데이터를 고객 ID 단위로 그룹화하고, 해당 고객의 총 구매 금액을 계산합니다.
3. intermediate 서버: LEAF 노드에서 계산된 고객별 총 구매 금액을 합칩니다.
4. root 서버: intermediate 서버에서 올라온 모든 값을 합치면서 총 구매 금액을 기준으로 소팅합니다. 소팅이 끝난 후에, 상위 5명의 레코드를 반환합니다.

마무리

지금까지 MySQL, MongoDB, BigQuery의 아키텍처와 구동원리를 알아보았습니다. 모두 동일한 관점에서 글을 작성하지는 않았지만, 글을 통해 각 DB/DW가 가지고 있는 특징을 이해한다면 훨씬 효과적인 DB 선택과 그에 따른 효율적인 아키텍쳐를 구성할 수 있을 것입니다.

참고문헌

• Real MySQL 8.0 written by 백은빈, 이성욱
• mongoDB Story 2: mongoDB 특징과 구성요소 : NHN Cloud Meetup
• 카카오와 MongoDB
• MongoDB 서버 구축 및 아키텍쳐 - 엘키의 주절 주절
• MongoDB 부수기
• BigQuery 개요  |  Google Cloud
• Dremel: Interactive Analysis of Web-Scale Datasets (Google paper)
• 갈아먹는 BigQuery [1] 빅쿼리 소개
• 갈아먹는 BigQuery [2] 빅쿼리 스키마 및 데이터 모델
• 구글 빅데이타 플랫폼 빅쿼리 아키텍쳐 소개

데이터 스트리밍의 중요성

데이터 스트리밍은 데이터가 생성되는 즉시 지속적으로 데이터를 수집, 처리, 분석하는 기술을 말합니다. 해당 방법은 전통적인 배치 처리 방식과 대비되며, 실시간으로 정보를 파악하고 즉각적인 피드백 루프를 생성할 수 있다는 점이 가장 큰 장점입니다. 아래는 스트리밍 데이터 처리가 쓰이고 있는 예시입니다.

예를 들어, 금융 서비스 업계에서는 주식 시장의 변동을 실시간으로 감지하고 자동으로 거래를 실행하는 알고리즘 거래 시스템이 필요합니다. 또한, 전자상거래 플랫폼은 사용자의 클릭스트림 데이터를 실시간으로 분석하여 개인화된 상품 추천을 제공해야 합니다.

이처럼 최신에 들어서는, 웬만한 기업은 서비스의 원활한 운영과 실시간 데이터 분석을 위해 스트리밍 데이터를 처리하려고 노력하고, Kafka나 Flink와 같은 프레임워크를 하나 이상 채용하여 사용하고 있습니다.

본 글의 말미에는 가장 인기있는 스트리밍 데이터 처리 프레임워크인 Kafka 를 소개할 예정입니다. 그 전에, 스트리밍 데이터에 대한 이해를 위해 필요한 주요 개념들 (ex: EDA, Topic, Micro-batch, etc)을 먼저 설명하려고 합니다.

가장 먼저 스트리밍 데이터 플랫폼을 알아보기 위해 필요한 개념은 이벤트 기반 아키텍처 (Event-Driven-Architecture) 입니다.

이벤트 기반 아키텍처 (Event-Driven-Architecture)

비즈니스는 수도 없이 많은 동적인 사건들의 발생으로 이루어져 있습니다. 유저의 회원가입부터 로그인, 장바구니 담기, 상품 구매, 상품 재구매 까지 모두 이벤트의 일종이라고 볼 수 있습니다.

이벤트 기반 워크플로우에서는 데이터 엔지니어링 수명 주기의 다양한 부분에서 이벤트를 생성, 라우팅, 소비의 프로세스가 진행됩니다.

이벤트 기반 아키텍처에서는 위에서 설명한 워크플로우를 기반으로 다양한 서비스간 통신을 진행합니다.

출처: https://akasai.space/architecture/about_event_driven_architecture/

해당 아키텍처의 장점은 이벤트의 상태를 여러 서비스에 분산시키기 때문에, 오프라인 상태가 되거나, 분산 시스템에서 노드에 장애가 발생하거나, 여러 소비자 또는 서비스가 동일한 이벤트에 접근하도록 할 때 유용하다는 점입니다. 보통, 서비스가 느슨하게 결합된 경우에는 항상 이벤트 중심 아키텍처를 포함합니다.

이벤트 기반 아키텍처는 아래와 같은 이유로 인기가 더욱 높아지고 있습니다.

1. 이벤트 기반 아키텍처의 핵심 계층인 메시지 큐와 이벤트 스트리밍 플랫폼은 클라우드 환경에서 더 쉽게 설정하고 관리 가능
2. 실시간 분석을 직접 통합하는 어플리케이션인 데이터 앱의 증가

핵심적으로, 이벤트 기반 아키텍처는 이벤트가 어플리케이션 작업을 트리거하고 실시간에 가까운 분석을 제공할 수 있습니다.

또한, 데이터 수집과 변환 단계에서도 원천 시스템에서 메시지 전달을 위해 사용했던 것과 같은 이벤트 스트리밍 플랫폼을 사용하여 실시간 분석을 위한 데이터를 처리할 수 있다는 점이 주요합니다.

메시지 큐와 스트리밍 플랫폼

이벤트 기반 아키텍처와 관련하여 메시지 큐와 스트리밍 플랫폼이라는 용어가 있는데, 혼용되는 경우가 많습니다.

메시지 & 메시지 큐

먼저, 메시지는 이벤트 기반 시스템에서 불연속적이고 단일한 신호의 일종으로, 둘 이상의 시스템 간에 전달되는 원시 데이터입니다.

메시지 큐는 게시 및 구독 모델을 사용하여 개별 시스템 간에 데이터를 비동기적으로 전송하는 메커니즘입니다.

기본적으로, 데이터는 메시지 큐에 게시되어 1명 이상의 구독자에게 전달되고, 메시지가 수신되면 큐에서 삭제됩니다. 메시지 큐를 사용하면 어플리케이션과 시스템을 서로 분리할 수 있어, 보통 MSA 환경에서 많이 사용됩니다.

메시지 큐는 메시지를 버퍼링해 일시적인 부하 급증을 처리하고, 복제 기능을 갖춘 분산 아키텍처를 통해 메시지를 내구성 있게 보존합니다. 메시지 큐에서는 메시지 순서 지정, 전달 빈도에 대한 개념이 중요합니다.

메시지 순서 지정

메시지가 생성, 전송, 수신되는 순서는 다운스트림 사용자에게 큰 영향을 미칠 수 있는데, 사실 분산 메시지 큐의 순서는 까다로운 문제입니다.

메시지 큐는 종종 모호한 순서와 선입선출 (FIFO) 개념을 적용합니다. 당연히 엄격하게 FIFO가 적용되면 먼저 들어온 메시지가 후에 들어온 메시지보다 먼저 수신될테지만, 고도로 분산된 시스템에서 잘못된 순서로 게시되거나 수신될 수 있습니다.

당연한 말이겠지만, 위 문제는메시지 큐 기술이 순서를 보증해줘야만 해결해 줄 수 있다. (ex: AWS SQS 표준 큐의 오버헤드 관리)

전달 빈도

메시지는 정확히 한번 발송되면 사용자가 메시지를 확인한 뒤 메시지는 사라지며 다시 전달되지 않고, 적어도 한번 송신된 메시지는 여러 명의 유저 또는 같은 유저가 2회 이상 소비할 수 있습니다.

따라서, 사용자가 메시지를 완전히 처리했지만, 처리를 확인하기 전에 실패할 때에 대한 대응을 적절히 할 수 있을 것입니다.

이상적으로는 시스템이 멱등성 상태여야 하고, 그러한 상태라면 메시지를 여러 번 처리한 결과와 한번 처리한 결과는 같을 것입니다.

스트림 & 스트리밍 플랫폼

이벤트는 ‘일반적으로 어떤 상태의 변화와 같은 무언가가 발생한 것’이고 단일 이벤트는 key, value, timestamp 와 같은 특성을 포함합니다.

스트림은 이벤트 레코드의 추가 전용 로그입니다. 이벤트가 발생하면 순서대로 누적되며, timestamp 또는 id로 이벤트 순서를 정렬할 수 있고, 여러 이벤트에 걸쳐 무슨 일이 일어났는지를 살펴볼 때 스트림을 사용할 수 있습니다.

메시지와 스트림은 pub → sub구조로 메시지를 전달한다는 점에서 유사하지만, 가장 큰 차이는 메시지 큐가 주로 특정 전달을 보장하는 메시지 라우팅에 사용된다는 것이다. 이벤트 스트리밍 플랫폼은 정렬된 레코드 로그에서 데이터를 수집하고 처리하는데 사용됩니다.

스트림의 추가 전용 특성으로 인해 레코드는 장기적인 보존 기간에 걸쳐 유지되므로, 여러 레코드의 집계 또는 스트림 내 특정 시점으로 rollback 기능과 같은 레코드의 복잡한 프로세스를 실행할 수 있습니다.

스트림을 처리하는 시스템은 메시지를 처리할 수 있으며, 스트리밍 플랫폼은 메시지 전달에 자주 사용됩니다. 또한, 위에서 언급했다시피 메시지 분석을 수행할 때는 메시지를 스트림에 축적하고, 나중에 특정 값에 대한 추세와 통계를 확인할 수 있습니다.

이제부터는, 이벤트 스트리밍 플랫폼에서 몇 가지 중요한 특성에 대해 설명하려고 합니다. Kafka를 포함해서 거의 모든 실시간 처리 프레임워크에서도 동일한 특성과 개념이 통용됩니다.

토픽 (Topic)

스트리밍 플랫폼에서 생산자는 관련 이벤트 모음인 토픽에 이벤트를 스트리밍합니다.

사실, 토픽을 여러 개 두거나 하나의 토픽에 여러 생산자를 할당하거나 하는 제한은 없고 보통 개발자가 환경에 맞게 설정하면 됩니다. 나의 경우로는, 암호화폐 거래소 별로 비트코인의 시장가를 포함한 여러 정보를 Kafka를 이용하여 실시간 처리하는 프로젝트를 진행하였는데 (엄밀히 말하면 마이크로 배치ㅎ), 거래소 별로 토픽을 할당하였다.

‘견고한 데이터 엔지니어링’ 책에서는 토픽 개념을 설명하기 위해서 일종의 주문처리 시스템을 가정하고, web orders라는 토픽, marketing과 fulfillment(주문처리)라는 생산자를 설정합니다.

fulfillment 구독자 (sub)은 이벤트를 사용해 주문 처리 프로세스를 트리거하고, marketing은 실시간 분석을 실행하거나 마케팅 캠페인을 조정하기 위해 ML 모델을 학습하고 실행할 것입니다. 글 말미에 추가로 설명하겠지만, 현업에서 위와 같은 상황이라면 보통 web orders 토픽 생산자를 Kafka, marketing을 processing하는 역할로 Spark Streaming로 처리하는 사례가 많은 것 같다.

스트림 파티션 (Stream Partition)

스트림 파티션은 스트림을 여러 스트림으로 분할(partition)한 것입니다.

메시지는 파티션 키에 따라 파티션 간에 분산되고, 파티션 키가 같은 메시지는 항상 같은 파티션에 저장됩니다. 일종의 MongoDB의 해쉬 인덱스나 해쉬 테이블처럼, 함께 처리해야 할 메시지끼리 동일한 파티션 키를 설정하여 분산시킨다고 생각하면 됩니다.

그러나, 우리는 이때 자연스럽게 특정 파티션에 메시지가 몰리는 상황이 아마 떠오를 것입니다. 이것을 hotspotting현상 이라고 합니다. 파티션 하나에 전달되는 메시지의 수가 불균형한 현상인데, 파티션 키를 적절하게 조절하며 해당 현상을 방지해야할 것이다.

내결함성과 복원성

이벤트 스트리밍 플랫폼은 일반적으로 다양한 노드에 스트림이 저장되는 분산형 시스템입니다. 따라서, 노드가 다운되면 다른 노드가 해당 노드를 대체해 스트림에 계속 접근 할 수 있는데, ‘견고한 데이터 엔지니어링’ 책에서는 이러한 특성을 내결함성과 복원성이라고 칭합니다. 이러한 특성 때문에, 이벤트 스트리밍 플랫폼은 이벤트 데이터를 안정적으로 생성, 저장 및 수집할 수 있는 시스템이 필요할 때 좋은 선택이 될 수 있습니다.

스트리밍 스토리지

최초의 실시간 쿼리 엔진은 어떻게 보면 OLTP기반의 트랜잭션 데이터베이스라고 볼 수 있다. 하지만, 대량에 데이터에 걸쳐 실행되는 분석 쿼리의 경우 확장 및 잠금의 제한으로 대용량에 적합한 쿼리를 실행시키기 어렵습니다.

기본적으로 스트리밍 데이터는 배치성 데이터와 스토리지 요구 사항이 다릅니다. (배치 데이터에 대한 스토리지 개념은 DW/DL, HDFS, Spark, iceberg에 대한 내용과 결합하여 다른 포스팅을 통해 소개드리려고 합니다 🙂)

메시지 큐의 경우, 저장된 데이터는 일시적이며 일정기간이 지나면 사라질 것으로 예상되지만, Kafka와 같이 분산되고 확장 가능한 스트리밍 프레임워크는 매우 오랜 기간 동안 스트리밍 데이터를 보존할 수 있습니다.

Kafka는 자주 접근하지 않는 오래된 메시지를 객체 스토리지에 푸쉬해 무기한 데이터 보존을 지원하는데, 이러한 기능은 AWS Kinesis나 GCP Pub/Sub도 지원한다.

위와 같은 데이터 스토리지 특성은 리플레이 개념과 연관되어 있습니다.

리플레이

리플레이는 스트리밍 스토리지 시스템의 표준 데이터 검색 메커니즘입니다.

리플레이를 사용하면 스트리밍 시스템에 저장된 과거 데이터의 범위를 반환할 수 있기 때문에, 시간범위에 걸쳐 배치 쿼리를 실행하거나 스트리밍 파이프라인에서 데이터를 재처리하는 데 사용할 수 있습니다.

Kafka와 AWS Kinesis나 GCP Pub/Sub은 모두 이벤트 보존 및 리플레이를 지원하지만, RabbitMQ 같은 경우에는 일반적으로 모든 사용자가 메시지를 소비한 후 메시지를 삭제합니다.

Stream-to-Batch 스토리지 아키텍처

Stream-to-Batch 스토리지 아키텍처는 람다 아키텍처와 유사점이 있지만, 해당 아키텍처는 기본적으로 스트리밍 스토리지 시스템의 토픽을 통과하는 데이터는 여러 소비자에게 기록되는 형태입니다.

이러한 소비자 중 일부는 스트림에 대한 통계를 생성하는 실시간 처리 시스템일 것입니다.

배치 데이터의 흐름은 아래와 같을 것입니다. 배치 스토리지 사용자는 장기 보전 및 배치 쿼리를 위해 데이터를 쓸 것이고 배치 소비자는 시간이나 배치의 크기에 대한 설정 가능한 트리거에 근거해 S3 객체를 생성할 수 있는 AWS Kinesis Firehose와 같은 시스템이 될 수 있습니다.

BigQuery와 같은 시스템은 스트리밍 데이터를 스트리밍 버퍼로 수집합니다. 이 스트리밍 버퍼는 자동으로 컬럼형 객체 스토리지로 다시 초기화 될 것이고, 쿼리 엔진은 스트리밍 버퍼와 객체 데이터 모두에 대한 원활한 쿼리를 지원해 사용자에게 거의 실시간에 가까운 최신 테이블 뷰를 제공할 수 있습니다.

스트리밍 데이터 처리 시 고려해야 할 사항

메시지와 스트림은 늘 유동적

스트리밍 데이터에 대한 수집은 데이터의 게시, 소비, 재게시, 재소비와 함께 비선형적일 수 있습니다. 따라서, data flow를 충분히 고려하여 실시간 데이터 파이프라인을 구성해야합니다.

또한, 대부분의 데이터의 형태가 JSON과 같은 반정형 구조일 확률이 높으므로, 페이로드의 스키마가 즉흥적으로 변경될 수 있습니다. 생산자 쪽에서 새로운 필드를 도입한 구조의 데이터를 송신하면, 대상이 되는 DW나 처리 파이프라인이 인식하지 못하는 상황은 좋지 않기 때문에 유연한 스키마를 유지하는 것이 중요합니다.

추가로, 후에 설명할 CDC 시스템의 경우에도 필드를 다른 타입으로 (예: 국제표준화기구의 날짜/시간 datetime형식 대신 문자열로) 리캐스팅하는 이슈가 존재할 수 있다.

처리량에 대한 모니터링

메시지와 이벤트는 가능한 짧은 지연시간 (latency)를 통해 흐르게 해야 합니다. 즉, 적절한 파티션이나 샤드에 대한 대폭과 처리량을 프로비저닝 할 줄 알아야 합니다.

그렇기에, 이벤트 처리에 충분한 메모리, 디스크 및 CPU 자원을 제공하고, 실시간 파이프라인을 관리할 때는 자동 계산 기능을 사용해 트래픽의 급상승에 대처하고 부하 감소에 따른 비용도 절감해야 합니다. 때문에 스트리밍 플랫폼 관리에는 배치 플랫폼보다 오버헤드가 발생할 여지가 많습니다.

마이크로 배치 vs 진짜 Real-time

요즘말로 하면 ‘성능이 좋은 서버에서 실행되는 REST API는 소켓 통신과 구별할 수 없다’ 이런 느낌일까?

사실 우리가 ‘실시간’ 이라고 칭했던 것들이 대부분 ‘마이크로 배치’ 형태였을 가능성이 높습니다.

마이크로 배치는 배치 지향 프레임워크를 스트리밍 상황에 적용하는 방법으로, 2분 간격에서 1초 간격까지 실행할 수 있습니다. Spark Streaming과 같은 것이 대표적인 마이크로 배치 프레임워크라고 볼 수 있는데, 높은 배치 빈도로 자원을 적절히 할당하면 실시간과 유사한 성능을 발휘할 수 있습니다.

아래 사진은 DataFrame을 기반으로 스트리밍 처리하는 Structured Streaming의 예시입니다.

진정한 Real-time을 구현하는 스트리밍 시스템인 Flink Streaming은 하나의 이벤트를 처리하도록 설계되었습니다. (물론 exactly once 전달은 Spark Streaming도 동일하게 적용되기는 한다) 그렇기에, 오베헤드가 상당할 것으로 쉽게 예상할 수 있습니다.

그러나, Flink Streaming의 경우에도 개별 이벤트에 데이터를 추가하는 기본적인 강화 프로세스에서는 지연시간이 짧은 이벤트를 한 번에 하나씩 전달할 수 있기 때문에 역설적이게도 여전히 많은 배치 프로세스가 발생합니다.

그래서 사실 당연히 정답이 없을 뿐더러, 개인적인 생각으로는 두 기술을 엄격하게 구별시키는 것도 큰 의미는 없을 것 같습니다. _(여담으로, 마이크로 배치라는 용어 자체가 경쟁기술을 배제하기 위해 사용되기도 하였다고 한다.) _ 결론적으로, 각자의 환경에 맞게 도메인 지식이 어느정도 받쳐주는지를 고려하면서 시스템의 요구사항을 적절하게 해소하면 된다고 생각한다 :)

변경 데이터 캡처 (Changed Data Capture)

사실, 스트림에는 이벤트 스트림과 CDC라는 두 가지 주요 유형이 존재한다.

CDC는 데이터베이스에서 발생하는 각 변경 이벤트(예: CRUD)를 추출하는 방법으로, DB 간에 거의 실시간으로 복제하거나 다운스트림 처리를 위한 이벤트 스트림을 생성하는 데 자주 사용됩니다.

흔히, 요즘 데이터엔지니어링 분야에서 많이 언급되는 트렌드인 Zero-ETL을 구축한다고 하면, CDC가 기술적 배경이 될 확률이 높습니다.

CDC는 DB의 종류에 따라 다르게 처리됩니다.

RDBMS의 경우는 스트림을 생성하기 위해 처리될 수 있는 이벤트 로그를 종종 생성해 DB 서버에 직접 저장하고, 많은 클라우드 NoSQL의 경우는 로그 또는 이벤트 스트림을 목표로 하는 스토리지 위치로 전송할 수 있습니다.

많은 CDC의 형태가 있지만, 스트리밍 데이터의 관점에서 볼 때는 연속 CDC 개념을 살펴보아야 합니다.

연속 CDC

연속 CDC는 모든 테이블 이력을 캡처해 실시간 데이터베이스 복제 또는 실시간 스트리밍 분석을 위한 거의 실시간 데이터 수집을 지원합니다. 일반적으로 연속 CDC는 정기적인 쿼리를 실행해 테이블 변경사항을 일괄적으로 가져오는 것이 아니라, 데이터베이스에 대한 각 쓰기를 이벤트로 처리합니다.

OLTP기반의 RDBMS를 대상으로 연속 CDC 이벤트 스트림을 캡쳐하기 위해 사용하는 것은 보통 로그 기반 CDC 입니다.

출처: https://www.striim.com/blog/log-based-change-data-capture/

데이터베이스 이진 로그는 데이터베이스의 모든 변경 사항을 순차적으로 기록하기 때문에, CDC 도구는 이 로그를 읽고 이벤트 형식으로 Debezium과 같은 플랫폼을 타깃으로 전송하게 됩니다. (여기서 말하는 이진 로그는 우리가 MySQL에서 replication을 진행할 때 사용하는 바이너리 로그 기반 복제의 이진로그와 동일한 개념입니다)

이제 기본적인 스트리밍 데이터에 대한 이해가 끝났으니, 이벤트 스트리밍 플랫폼으로 가장 많은 사랑을 받고 있는 Apache Kafka에 대해 본격적으로 알아보자.

Why Kafka?

보통 파이프라인의 성능은 두 가지로 결정됩니다. 바로 latency와 처리량 입니다. 아무래도 스트리밍 데이터를 처리하는 프레임워크는 그 성격 상 latency에 치중될 수 밖에 없습니다. 그러나, Kafka는 디스크 기반의 로깅 메커니즘을 통해 처리량 또한 준수한 수준을 보입니다.

특히, Kafka 는 강력한 데이터 복제를 ** ** 브로커를 통해 지원합니다.

브로커는 Kafka 클러스터를 구성하는 기본 단위입니다. Kafka 에서는 데이터를 여러 브로커에 복제하여 저장하여, 어떤 브로커에 문제가 발생해도 데이터 손실 없이 처리를 계속할 수 있음을 의미하며, 실시간 시스템에서는 이러한 내구성이 매우 중요한 요소로 작용합니다.

추가로, 분산 시스템으로 설계되어 있어 데이터를 여러 서버(브로커)에 걸쳐 저장하고 처리할 수 있기에, 단일 노드의 성능 한계를 넘어서는 확장성과 높은 처리량을 가능하게 합니다.

이외에도, Kafka는 다양한 프로그래밍 언어 및 프레임워크와의 호환성, 그리고 이미 형성되어 있는 대규모 커뮤니티 지원을 통해 많은 기업들이 Kafka를 채용하여 비즈니스 데이터를 효율적으로 관리하고 있습니다.

Kafka Architecture

Kafka config: Topic, Partition, Replica, Leader, Follower…. etc

토픽과 파티션은 위에서 이미 언급했지만, 조금 더 상세한 이해를 위해 구체적으로 한번 더 짚고 넘어가겠다.

토픽 (Topic)

토픽은 데이터의 카테고리나 분류를 나타내는 단위입니다. 예를 들어, "user_signups" 또는 "order_transactions"와 같은 토픽을 만들 수 있습니다. 각 토픽은 하나 이상의 파티션을 가질 수 있습니다.

파티션 (Partition)

파티션은 토픽의 데이터를 분할하는 단위입니다. 토픽의 데이터가 여러 파티션에 나누어 저장되어 사용되기 때문에, 병렬 처리를 통해 높은 처리량을 달성할 수 있습니다. 처음, 해당 개념을 접했을 때 혼자 아래와 같은 예시를 만들어 이해하려고 애썼던 기억이 있다.

예를 들어, "user_signups"라는 토픽이 3개의 파티션으로 구성되어 있다면:

• 파티션 1: 사용자 가입 데이터의 1/3

• 파티션 2: 사용자 가입 데이터의 1/3
• 파티션 3: 사용자 가입 데이터의 1/3

이렇게 각 파티션은 토픽의 전체 데이터 중 일부를 독립적으로 저장합니다.

프로듀서가 토픽에 메시지를 쓸 때, Kafka는 해당 메시지를 특정 파티션에 할당합니다. 이 할당은 여러 가지 방법(예: 라운드 로빈, 키 해싱 등)으로 수행될 수 있습니다. 따라서, 기본적으로 토픽의 모든 데이터를 조회하려면 모든 파티션에서 데이터를 읽어와야 합니다. 단, 각 파티션 내에서의 메시지 순서는 유지되지만, 토픽 전체의 파티션 간 메시지 순서는 보장되지 않을 수 있습니다.

복제본 (Replica)

복제본은 파티션의 복사본으로, 데이터의 내구성과 고가용성을 보장하기 위해 사용됩니다.

위에서 언급했다시피, 만약 한 브로커가 실패하더라도 해당 브로커에 저장된 파티션의 복제본이 다른 브로커에 존재하기 때문에 데이터 손실을 방지할 수 있습니다.

Segment

Segment는 브로커의 로컬 스토리지에 저장되는 파티션의 물리적인 저장 단위입니다.

1. 파일 기반 저장: 각 세그먼트는 실제로 두 가지 주요 파일로 구성됩니다. 하나는 실제 메시지를 저장하는 .log 파일, 그리고 메시지의 위치를 빠르게 찾기 위한 .index 파일입니다.
2. Rolling: 세그먼트는 설정된 크기나 기간에 도달하면 "roll" 되며, 새로운 세그먼트 파일이 생성됩니다. 이것은 오래된 데이터를 효율적으로 삭제하거나 관리할 수 있게 해주는 장점이 있습니다.
3. 데이터 삭제 및 보존: 오래된 데이터를 삭제할 때, Kafka는 전체 세그먼트 파일을 삭제함으로써 효율성을 유지합니다. 데이터 보존 정책(retention policy)에 따라, 메시지가 보존되는 시간이나 세그먼트의 크기에 도달하면 해당 세그먼트는 삭제될 수 있습니다.
4. 효율적인 읽기 및 쓰기: 세그먼트 구조는 Kafka가 대량의 데이터를 빠르게 읽고 쓰는 데 있어 효율적입니다. 특히, 순차적인 디스크 I/O 작업은 랜덤 액세스보다 훨씬 빠르기 때문에 세그먼트는 이러한 순차적인 접근의 이점을 활용합니다.

각 파티션은 리더(leader)와 하나 이상의 팔로워(follower)로 구성

1. 쓰기 작업: 프로듀서가 데이터를 쓰려고 할 때, 이 작업은 해당 파티션의 리더에게 전달됩니다. 오직 리더만이 해당 파티션에 데이터를 쓸 수 있습니다.
2. 팔로워 동기화: 리더가 데이터를 받고 기록한 후, 이 데이터는 팔로워들에게 복제됩니다. 팔로워는 리더로부터 데이터를 주기적으로 가져와서 자신의 로그에 동기화합니다. 이렇게 해서, 만약 리더가 실패하면, 팔로워 중 하나가 새로운 리더로 승격될 수 있으며, 데이터 손실 없이 서비스를 계속 제공할 수 있습니다.
3. 읽기 작업: 컨슈머가 데이터를 읽을 때 기본적으로 리더에서 데이터를 읽습니다. 그러나 Kafka 설정에 따라, 컨슈머가 팔로워 브로커로부터 직접 데이터를 읽는 것도 가능하며, 이를 통해 읽기 처리량을 분산시킬 수 있습니다.

참고문헌

• 데이터 스트리밍: 장점, 예시 및 사용 사례 | KR
• 견고한 데이터 엔지니어링 (written by Joe Reis & Matt Housley)
• Apache Kafka 공식문서
• 데이터 엔지니어링 데브코스 lecture note (created by 한기용)

본 Git으로 개발 생산성 올리기 시리즈는 현업에서 Git을 사용하며 가장 유용하게 사용했던 명령어 및 개발 생산성을 올릴 수 있었던 경험을 공유하기 위해 제작되었다. 0편에서는 본격적인 시리즈에 들어가기 전, 기본적인 Git에 대한 간단한 소개 및 동작원리, 그리고 Git에 관한 흥미로운 글을 소개해보려고 한다. Git을 사용한 경험이 한번이라도 있다면, 충분히 이해하실 것으로 기대합니다 :)

1. Git이란?

Git은 분산 버전 관리 시스템(DVCS)으로, 소프트웨어 개발 프로젝트에서 소스 코드의 변경 사항을 추적하고 여러 사용자 간의 작업을 조율하는 데 사용되고 있다. 분산 버전 관리 시스템인 이유는, 원본 코드는 원격 레포지토리에 공유되어 있고, 각자의 로컬 환경에 로컬 코드가 분산되어 있기 때문이다. 참고로, Git은 Linux를 만든 리누스 토르발스에 의해 2005년에 개발되었다.

Git을 왜 써야할까?

Git을 사용하는 주된 이유는 팀 내 또는 개인 프로젝트에서의 소스 코드 관리를 용이하게 하기 위함이다. Git은 변경 사항을 효율적으로 추적하고, 버전을 관리함으로써 여러 버전의 문서를 안전하게 보관하고 필요시 원하는 버전으로 쉽게 되돌아갈 수 있게 한다.

SVN과의 차이점

SVN은 중앙집중식 버전 관리 시스템(CVCS)이며, 모든 파일의 버전 관리 정보가 중앙 서버에 저장됩니다. 위에서 언급한 바와 같이 Git은 분산 버전 관리 시스템으로, 각 사용자가 전체 저장소의 복사본을 로컬에 보유하며, 이로 인해 네트워크 상태에 구애받지 않고 작업을 진행할 수 있다. 브랜칭이나 병합과 같은 유연성도 우수하다.

2. Git의 기본적인 동작 원리

• 구성
• 동작원리
  • Commit은 어떻게 이루어질까?
  • 충돌이 왜 이러날까?

구성

• Working Directory: 실제 파일들이 위치한 디렉토리로, 사용자가 현재 작업하고 있는 공간이다.

• Staging Area: 커밋하기 전의 준비 영역으로, Git에서 변경사항을 임시로 저장하는 곳이다. ex) git add를 사용하여 파일 변화를 스테이지 위에 올린다

• Local Repository: 사용자의 PC에 위치한 저장소로, 프로젝트의 모든 버전 정보를 포함합니다. ex) git commit을 하여 스테이지 위의 파일 변화들을 로컬 레포지토리에 기록한다

• HEAD: 현재 체크아웃된 커밋을 가리키는 포인터로, 가장 최근의 작업 상태를 나타낸다. 보통 커밋 기록을 볼 때, _HEAD 0XEF56 _ 와 같은 로그를 볼 수 있는데, 현재 0XEF56 라는 커밋을 HEAD가 가리키고 있고, 이가 최근 커밋이라는 것라고 생각하면 된다. 후에, cherry-pick이나 해당 커밋으로 돌아가기 위해서는 해당 값(0XEF56)을 기반으로 진행되게 된다.

• Remote Repository: 로컬과 분리되어 있는 Git원격 저장소 ex) git push를 하여 원격저장소로 커밋 결과들을 올린다

동작 원리

사실 Git을 처음 사용하기 시작했을 때, 많은 고민을 하지 않았었다. add, commit, push, pull만 잘 동작하면, 크게 무리가 없을 것 같다고 생각했고 실제로도 필요가(?) 없었다. 큰 장벽을 만나기 전까지.... 그래서 한번 쯤은, 하루 이틀 정도 Git의 작동원리나 사용사례에 대한 자료들을 찾아보며 깊게 이해해 보시길 추천드린다.

사실 기본적인 동작원리는 위 구성 파트에서 설명이 다 이루어졌다. 따라서, 동작원리와 별개로 Git을 이해하며 개인적으로 중요하게 생각했던 개념 몇 가지를 설명하려고 한다.

Commit은 어떻게 이루어질까?

위에서는 하루 이틀이라고 칭했지만, 나는 사실 꽤 오랜기간 Git에 관하여 제대로 이해하지 못했다. 지금도 완벽하다고 생각하진 않고, 누가 설명을 요청하면 아직도 헷갈리는 개념이 많아 쉽지 않다. 하지만, 기본적인 이해로 넘어가는 허들을 극복하는 순간은 Commit에 대한 이해를 했을 때였다.

기본적으로, 버젼관리에서의 핵심은 파일 변화이다. 수정, 삭제, 추가 등의 행위를 말하는 것이다. Git에서는 이것을 측정 및 감지하는 단위가 Commit인 것이다.

앞서 말한 SVN은 이전 파일과의 차이를 추적하며 동작이 되는 구조인데, Git은 차이 뿐만 아니라 전체 코드에 대한 기록도 유지된다. 이는 용량적인 문제가 발생하는 것처럼 보이지만, SVN은 처음 작업이 시작할 때부터의 변경사항을 모두 추적해야 하므로 불필요한 연산이 발생한다. Git의 커밋 개념으로 비교하자면, 처음 커밋부터 모두 체크해야한다는 것이다. 허나, Git은 알다시피 직전 커밋과의 비교만 진행된다. 그리고, 파일 변화가 이루어지지 않은 작업물들은 일종의 심볼릭 링크 형태로 저장되기 때문에, 용량적인 문제를 야기하지 않는다.

정리하자면 Git의 버전관리는 이전 커밋과의 차이점만 비교하며 이루어지고, 일일이 코드의 차이점을 연산하지 않고 스냅샷(=커밋)을 찍어 비교한다.

무식(?)하지만 효율적인 방식으로, 오직 커밋 단위로 Git은 동작하기 때문에 이에 대한 이해가 필수적이라고 생각한다.

충돌이 왜 이러날까?

아마 Git이 우리의 머리를 아프게 하는 순간은 병합을 시도했을 때 충돌이 일어났을 때일 것이다. 충돌이 일어나는 이유는 간단하지만, 해결법은 다양할 것이다. 예를 들어, 커밋을 되돌리거나, 충돌난 부분을 직접 수정하거나, 극단적(?)으로 force 명령어를 사용하는 것이다. 개인적으로, force 명령어는 개인만이 코드를 관리하지 않는 이상 추천하지 않고 직접 수정하는 것을 추천한다 :)

*일단 충돌이 일어나는 이유는, 쉽게 말하면 같은 코드를 수정했기 때문이다. *

일반적으로, 병합을 시도할 때 3가지 경우의 수가 존재한다. 병합을 시도하는 브랜치를 A와 B로 설명하려고 한다.

1. Fast-Forward: A (변경사항 a) + B (변경사항 X) = A
   • B는 변경사항이 없기 때문에, 그냥 A로 돌려감기 하면 됌
2. Merge-commit: A (변경사항 a) + B (변경사항 b) = A + B
   • 변경사항이 없기 때문에 병합이 정상적으로 진행되고, 병합되었다는 커밋이 기록 됌

충돌이 일어나는 경우는 3번째 경우이다. *A (변경사항 a) + B (변경사항 a') = 변경사항에 대한 코드가 겹치기 때문에 병합 못함 *

이런 생각을 할 수 있다. 코드 출처가 같긴 하지만, 서로의 코드에 영향이 가지 않고 기존 코드 수정 없이 코드가 추가 된 것인데도 충돌이 왜 발생해? 앞에서 설명했던 것과 같이 Git은 커밋을 단위로 변화를 감지하고 이는 스냅샷을 비교하는 것이다. 개행을 했든 주석을 추가했든, 변화는 변화일 뿐이다. 그러니, 침착하게(?) 충돌난 부분을 체크하고 직접 수정하는 것이 가장 빠른 해결책으로 생각된다 ㅎ 당연히, 충돌이 해결되면 2번의 경우와 같아지므로 Merge-commit이 발생한다.

아래는 Azure 기술 블로그에 병합 관련하여 포스팅된 글이다. 추가적으로 Git에 대한 이해를 더할 수 있는 좋은 글들이 많으니 참고 바랍니다 :) https://learn.microsoft.com/ko-kr/azure/devops/repos/git/merging?view=azure-devops&tabs=visual-studio-2022

Git pull은 나쁘다?

초기 Git 개발에 참여했던 Felipe Contreras가 Why is git pull broken?라는 자극적인 제목으로 작성한 글을 우연히 보게되었는데 인상깊어 소개합니다ㅎ

해당 글의 주된 내용은 git push와 git pull은 git 작동원리에서는 반대의 의미를 가지고 있지 않고, git pull의 지나친 사용을 지양해야 하는 이유를 상세하게 설명합니다. git pull을 습관적으로 사용하고 (바로 나..ㅎ) rebase 작동원리를 이해하고 싶은 분께 추천드립니다 :)

이전에 관련 세션을 준비했다가 삑난(?) 경험이 있어서 준비했던 내용에 대해 시간이 된다면 아래 포스팅을 추가로 올릴 예정입니다 :)

1. Git으로 개발 생산성 올리기 1편 (feat. git stash를 알고 삶의 질이 올라갔다?)
2. Git으로 개발 생산성 올리기 2편 (feat. 코딩 스타일 관리 자동화를 위한 Git Hook 도입기)
3. Git으로 개발 생산성 올리기 3편 (feat. Github Action with Terraform)

참고로, 관련해서 발표 준비를 위한 테스트 용도로 만들었던 git repository link는 다음과 같습니다. 후에 포스팅을 이어나간다면, repo는 cleanup해서 새로 생성해보겠습니다 ㅎ https://github.com/idle-danie/git_session

이 글에서는 기본적으로 OpenAI API(ChatGPT)를 활용하여 chatbot을 개발할 때, 중요한 API 2가지를 parameter 중심으로 풀어보고, 경험에서 얻은 Tip과 개인적인 견해를 소개해보려고 합니다. 참고로 각 API 제목에 공식문서 링크가 첨부되어 있습니다 :)

서론

개인적은 생각으로는, Attention is all you need 논문이 나온 뒤, LLM 모델에 대한 연구와 관심도가 폭발적으로 늘어났던 것 같다. 최근 몇년 간, 아카이브(인공지능 한정)는 90프로정도가 중국 연구원분들의 논문으로 채워져 있는 것 같다… 현업에서는 독자적인 LLM 모델을 만들거나 LLAMA와 같은 오픈소스를 활용하여 챗봇을 개발하려는 시도가 많아졌던 것 같다.

*OPENAI에서 ChatGPT4가 나오기 전까지는…. *

ChatGPT가 워낙 general하기도 하고, 답변 생성 성능이 타 모델에 비해 좋기도 해서 이제는 "OpenAI를 잘 활용하자"_로 많은 분들의 생각이 변한 것 같다. 물론, 연구비 지원이 가능한 기업들은 독자적인 모델을 지속적으로 개발 중이신 것 같긴하다. ~~_언젠가는 나도 기여하는 날이 왔으면...ㅎ~~ 다만, ChatGPT와 같이 general한 모델 보다는 특정 도메인에 customized된 챗봇을 개발하는 방향으로 산업이 변화하는 것으로 보인다. 이제 어떻게 OpenAI의 API를 사용하여, 나만의 customized chatbot을 제작할 수 있는지 알아보자.

추가로, 본 글에서 설명하는 API는 모두 OpenAI playground에서 미리 체험해볼 수 있으니 꼭 먼저 사용해보길 추천드립니다 :)

1. Text generation API

우리가 흔히 사용하는 웹 ChatGPT를 API로 만들었다고 생각하면 됩니다 :)

Code

공식문서에서 제공한 코드 예시는 아래와 같다.

MODEL = "gpt-3.5-turbo"
response = client.chat.completions.create(
    model=MODEL,
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain asynchronous programming in the style of the pirate Blackbeard."},
    ],
    """
    parameters
    ex) temperature=1,top_p=1, frequency_penalty=0, presence_penalty=0
    """
)
print(response.choices[0].message.content)

Parameter

message에 해당하는 prompt구성법은 굳이 언급하지 않겠다. 또한, prompt를 구성하는 절대적인 해답 또한 없다. 나중에 prompt-engineering은 따로 포스팅 하겠다.

API parameter가 여러가지 있지만, 아래 5개 정도가 개략적인 답변의 형태를 결정한다.

• N: 생성하는 답변 개수 (default = 1)
• Temperature: 값이 높을수록 창의성 & 무작위성 증가 (default = 1)
• TopP: 값이 높을수록 다양한 답변을 생성 (default = 1)
• Frequency Penalty: 값이 높을수록 반복을 더 많이 피하려고 함 (default = 0)
• Presence Penalty: 값이 높을수록 이미 언급된 내용을 피하려고 함 (default = 0)

모두 optional한 것이라 굳이 사용하지 않아도, OpenAI 에서 default로 설정한 값이 사용되도 큰 무리는 없을 것이다. (아마, 우리가 웹 ChatGPT에서 사용하는 값이 default값이 아닐까 생각한다)

*다만, 우리가 특정 도메인을 타겟하여 customize한다면 파라미터 운영을 해야하는 순간이 온다. *

예를들어, 어떤 도메인은 general하지 않은 즉 창의성이 돋보이는 답변이 생성되어야 할 수도 있고, 어떤 도메인은 outlier가 절대 나타나면 안되고 보편적인 답변이 우선시되어야 할 수 있다는 말이다. 따라서 해당 파라미터를 컨트롤하며, 특정 도메인에 맞는 최적해를 여러 실험을 통해 도출해야 한다. 막상 생각은 안나지만 이해를 돕기위해 도메인 예를 굳이 들어보자면, 전자는 '막장 드라마 대본 생성기', 후자는 _'공공기관 고객 응대 챗봇'_라고 생각할 수 있겠다.

N

N (응답 개수): 이 파라미터는 챗봇이 한 번에 생성하는 답변의 개수를 결정합니다. 높은 값은 더 많은 수의 다양한 답변을 생성하며, 낮은 값은 적은 수의 답변을 생성합니다.

위 코드에서 아래 line의 인덱스를 보고 의문을 가진다면, 아래의 N parameter 설명을 읽게 되면 이해하게 될 것이다.

print(response.choices[0].message.content)

우리가 웹에서 ChatGPT를 사용할 때, 우리는 답변을 1개 받는 이유는 N이 default값 1로 설정되어 있기 때문이다! 가끔, 우리가 답변 2개를 제시받고 그 중 더 선호하는 답변을 선택하라고 할 때가 있는데 그때는 N=2인 상태로 API response를 받는 것이다. 아마, 내부 로직 상에서 해당 답변을 선호된 답변으로 저장하고 아래 2가지 방향으로 develop되지 않을까? 라는 개인적인 생각이다.

1. 해당 chat session에 한하여, 선호된 답변의 방향성으로 prompt를 생성하도록 유도
2. 후에, 해당 모델을 선호된 답변을 바탕으로 학습모델 fine-tuning

*N 파라미터를 활용하는 방법은 2가지로 축약된다. *

1. 실험에 대한 검증: N=1로 설정했을 때, 실험 진행 중 생성된 답변에 대한 신뢰성이 보장할 수 없을 때 N값을 증가시켜 생성되는 답변의 양상을 지켜볼 수 있다.
2. 다양한 답변을 도출: 1번의 의도와는 다르게, 애초에 여러 prompt를 생성하고 싶을 수 있다. 다만, 최적 파라미터와 input prompt설정에 대한 실험이 완료되었을 때 추천한다 :) 그 이유는 N=n (n>1)일 때 경험 상, 전체적인 답변의 질이 낮아지는 점, 실험 결과를 객관적으로 분석하기 힘든 점 등이 있어 실험 단계에서는 위 의도를 관철시키는 것이 어렵기 때문이다.

*결국 우리의 목표는, N=1인 상태에서 최적의 답변이 생성되어야 한다는 것이다. *

Temperature & TopP

Temperature: 이 값은 챗봇의 응답이 얼마나 예측가능한지 또는 창의적인지를 조절합니다. 낮은 온도(예: 0.2)는 더 예측가능하고 일관된 답변을 생성합니다. 높은 온도(예: 2.0)는 더 창의적이고 예상치 못한 답변을 생성합니다. (range: 0 <= Temperature <=2)

TopP: 이 값은 챗봇이 고려하는 답변의 다양성을 조절합니다. 낮은 TopP 값은 챗봇이 더 일반적이고 보편적인 답변을 선택하게 하며, 높은 TopP 값은 더 다양하고 예측 불가능한 답변을 생성합니다.

공식문서에서의 TopP 설명을 번역하면 아래와 같다. 이 설정은 모델이 토큰을 선택할 때, 확률이 높은 상위 일정 비율의 토큰들만 고려하도록 합니다. 예를 들어, top_p가 0.1(10%)로 설정되어 있다면, 모델은 가능한 토큰들 중 확률 합이 상위 10%에 해당하는 토큰들만을 고려하여 다음 토큰을 선택하게 됩니다. 이 방법은 텍스트 생성에서 더 다양성을 부여하거나, 너무 예측 가능한 결과를 피하고자 할 때 유용합니다. 일반적으로 top_p 설정이나 temperature 설정 중 하나를 조정하는 것을 권장하지만, 두 설정을 동시에 조정하는 것은 권장하지 않습니다. 이는 두 설정 모두 생성 텍스트의 다양성과 예측 가능성을 조절하는 역할을 하기 때문입니다.

여기서 중요한 점은, 두 파라미터를 동시에 조작변인으로 설정하지 말아야 한다는 점이다! 물론, 정확하게 파라미터가 프롬프트에 작용하는 방식은 다르지만 중요한 것은 공식문서에서도 명시된 것처럼, 실제 실험을 진행할 시에 두 파라미터를 조작변인을 설정한다면, 답변의 창의성에 대한 명확한 수치를 평가하기 어려울 것이다. 그래서 추천하는 방식은, 아예 전체적인 파라미터 운영에서 두 파라미터 중 한 가지를 제외시키는 것이다. 이외에는 아래와 같은 2가지 방식으로 활용할 수 있을 것 같다.

1. Temperature vs TopP: 명확하게 아래와 같이 변인을 통제하여 둘 중 어떤 파라미터를 선택할지 선택 실험 A: Temperature=0.5, TopP=1.0 실험 B: Temperature=1.0, TopP=0.5
2. 여러 실험을 거쳐 (Temperature, TopP)에 대한 최적 조합 찾기

물론, 3번과 같이 여러 실험을 거쳐 조합에 대한 최적해를 찾는 것이 이상적이긴 합니다만 현실적으로 모델링을 직접 한 것이 아니기 때문에, 해당 파라미터가 실제로 어떤 작용을 하는지는 모른다는 점이 있습니다.

*결론: 하나는 default값 쓰고, 나머지를 조작하며 최적해 찾기 *

Frequency Penalty & Presence Penalty

Frequency Penalty (빈도 패널티): 이 설정은 챗봇이 반복되는 단어나 구문을 얼마나 피할지 결정합니다. 높은 값은 챗봇이 반복을 피하도록 하며, 낮은 값은 반복이 더 자주 발생하게 합니다. (range: -2.0 <= fp <= 2.0)

Presence Penalty (출현 패널티): 이 값은 챗봇이 이전에 사용한 내용을 얼마나 피할지 조절합니다. 높은 값은 챗봇이 같은 내용을 반복하지 않도록 하며, 낮은 값은 챗봇이 이미 언급한 내용을 다시 사용할 가능성이 높아집니다. (range: -2.0 <= pp <= 2.0)

두 parameter 모두 단어에 대한 중복 & 반복에 관한 논의이지만 구체적으로 어떤 차이점이 있는지에 대한 이해가 쉽지 않을 것 같아서 아래 코드와 예시를 통해 설명해보려 한다.

mu[j] -> mu[j] - c[j] * alpha_frequency - float(c[j] > 0) * alpha_presence

- Logits (mu[j]): Logit은 모델이 특정 단어를 선택할 확률을 나타내는 로그 확률입니다. 모델은 이러한 로그 확률을 사용하여 다음에 올 단어를 결정합니다.
- c[j]: 이는 모델이 현재 위치 이전에 특정 단어를 선택한 횟수를 나타냅니다.
- alpha_frequency (빈도 패널티 계수): 이 계수는 특정 단어가 반복될 때마다 그 단어의 로그 확률을 감소시키는 데 사용됩니다. 즉, 단어가 더 자주 나타날수록, 그 단어가 다시 선택될 가능성이 낮아집니다.
- alpha_presence (존재 패널티 계수): 이 계수는 단어가 한 번이라도 나타난 경우, 그 단어의 로그 확률을 일정량 감소시키는 데 사용됩니다. 즉, 단어가 이미 한 번 사용되었다면, 다시 나타날 가능성이 줄어듭니다.

예를들어 '고양이'라는 단어를 이미 한 번 사용했다면, '존재 패널티'는 '고양이'라는 단어가 다시 사용될 확률을 감소시키기 위해, 그 단어의 로그 확률에 직접적인 감소를 적용합니다. 만약 '고양이'가 여러 번 반복되었다면, '빈도 패널티'는 그 단어가 선택될 때마다 로그 확률을 더욱 감소시켜, 모델이 같은 단어를 계속해서 반복하는 것을 방지합니다.

결론적으로, 빈도 패널티는 **특정 단어가 반복될수록 점점 더 큰 패널티를 적용**하는 반면, 존재 패널티는 단어가 **문장에서 한 번이라도 나타나면 일정량의 패널티를 적용**합니다.

_위 설명이 어렵다면 아래 2가지 사항만 알고 넘어가자!
_
> 1. 빈도 패널티는 **특정 단어의 과도한 반복을 방지**하는 데 유용하며, 존재 패널티는 **다양한 단어와 구가 생성된 텍스트에 등장하도록 유도**하는 데 더 적합합니다.
2. 공식문서에서도 명시되어 있지만, 두 값 모두 1.0 이상으로 parameter 설정 시에는 답변의 퀄리티가 확실히 떨어지는 것으로 보입니다.

# 2. [Fine-tuning API](https://platform.openai.com/docs/guides/fine-tuning)

## Fine-tuning의 필요성
![](https://velog.velcdn.com/images/idle-danie/post/d3bb2b08-d9c2-4269-8752-ded5ce9ff838/image.png)

우리가 웹에서 사용하는 ChatGPT은 굉장히 general한 모델을 사용하고 있습니다. 
그렇기에, 어떤 질문에도 평균 이상의 답변을 생성해냅니다. 그러나, 각자 아래와 같은 경험이 있을 것입니다. 

특정 채팅 세션에서, 특정 보고서 형식에 맞게 질문 몇 가지를 던져주고 원하는 응답을 이끈다. 그리고, 일주일 뒤 다시 돌아와 해당 채팅을 다시 활용합니다. 나만의 모델을 제작한 셈입니다. 아래에서 말하는 개념과는 살짝 다르지만, **학습이 별도로 필요**하단 말을 하려고 하는 것입니다.

결론적으로, 특정 도메인에 맞춰 커스텀하기 위해서는 별도의 학습, **즉 fine-tuning이 필요합니다. **

fine-tuning은 머신러닝에서 **특정 작업에 최적화된 성능을 달성하기 위해 이미 훈련된 모델을 추가적으로 조정하는 과정**을 말합니다. 이 과정은 특히 딥러닝에서 널리 사용되며, 큰 데이터셋으로 사전에 훈련된 모델을 새로운, 종종 더 작은 데이터셋에 적용할 때 일반적으로 수행됩니다.

fine-tuning을 진행할 때는 기존 모델의 구조를 유지하면서 일부 매개변수만을 조정합니다. 이를 통해 모델은 새로운 데이터에 특화되어 성능이 개선되는 것을 목표로 합니다. 
예를 들어, 대규모 이미지 데이터셋으로 사전 훈련된 모델을 특정 동물을 분류하는 작업에 맞게 파인튜닝할 수 있습니다. 이 과정은 학습 시간을 크게 단축시키고, 필요한 데이터의 양을 줄이며, 전반적인 효율성을 높이는 효과가 있습니다.

일단 먼저 OpenAI에서 제공하는 **fine-tuning API를 사용하는 방법**부터 공식문서의 코드를 통해 알아봅시다.

## Code
### 1. Training data 제작하기

```python
from openai import OpenAI
client = OpenAI()

client.files.create(
  file=open("mydata.jsonl", "rb"),
  purpose="fine-tune"
)

Code에서 확인할 수 있듯이, 먼저 jsonl파일 형식으로 dataset을 준비해야 한다. 아래는 공식문서에서 제공하는 prompt 형식이다.

{"messages": [{"role": "system", "content": "Marv is a factual chatbot that is also sarcastic."}, {"role": "user", "content": "What's the capital of France?"}, {"role": "assistant", "content": "Paris, as if everyone doesn't know that already."}]}
{"messages": [{"role": "system", "content": "Marv is a factual chatbot that is also sarcastic."}, {"role": "user", "content": "Who wrote 'Romeo and Juliet'?"}, {"role": "assistant", "content": "Oh, just some guy named William Shakespeare. Ever heard of him?"}]}
{"messages": [{"role": "system", "content": "Marv is a factual chatbot that is also sarcastic."}, {"role": "user", "content": "How far is the Moon from Earth?"}, {"role": "assistant", "content": "Around 384,400 kilometers. Give or take a few, like that really matters."}]}

참고로, fine-tuning API에 활용될 수 있는 OpenAI 모델은 아래와 같습니다. gpt-3.5-turbo, babbage-002, davinci-002 모델마다 각각 지원하는 prompt형식이 다를 수 있습니다. 본 글은 가장 최신 모델인 gpt-3.5-turbo를 기준으로 설명하겠습니다.

gpt-4.0-turbo model이 작년 11월 DevDay 컨퍼런스를 통해 공개되고 이전 모델인 gpt-4.0을 fine-tuning 가능한 모델로 일부 파트너에게만 공개했습니다만 그 이후 업데이트 소식은 없습니다... 아마 ChatGPT웹에서 나만의 모델을 제작할 수 있는 gpt-store가 출시되고 이제 fine-tuning API는 관심밖으로...?

2. Fine-tuning 모델 제작

모델 제작시에 Parameter managing이 가능하다. 이 부분은 아래 Parameter section에서 제대로 설명하겠다. 일단 아래는 별다른 parameter를 넣지 않고 fine-tuning 작업을 생성하는 코드이다.

from openai import OpenAI
client = OpenAI()

client.fine_tuning.jobs.create(
  training_file="file-abc123", 
  model="gpt-3.5-turbo"
)

1번 과정에서 만든 training_file의 이름과 사용할 fine-tuning에 사용될 gpt 모델명을 명시해야 합니다. 소요시간은 추가로 학습시키는 데이터의 양에 따라 달라지고, 경험 상으로 prompt 10개 정도를 학습할 경우 5~10분 정도 소요됩니다.

참고로, 이미 fine-tuning으로 학습한 모델도 다시 fine-tuning을 진행할 수 있습니다! 이 경우, 기존에 제작한 모델명을 명시해주면 됩니다 :) ex) 별다른 네이밍 작업을 하지 않으면 ft:gpt-3.5-turbo:my-org:custom_suffix:id와 같은 스타일로 모델명이 할당됩니다

3. Fine-tuning 진행상황 추적

같은 OpenAI key(token)을 사용하며, 여러명이 작업을 할 경우 작업이 겹칠 수 있다. 같은 시간에 여러 작업이 병렬처리 되는 것이 아니라, 큐 형태로 먼저 시작된 작업부터 한번에 하나의 작업만 처리하게 되기 때문이다. 이 경우, 각자 상황에 맞게 해결하면 되지만 예상하다시피 아래 명령어들이 분명 필요해질 것 입니다 :) 현재는 처리 방식이 달라졌을 수 있습니다ㅎ

현재 진행중인 fine-tuning job list

client.fine_tuning.jobs.list(limit=10)

현재 fine-tuning job 상태 호출

client.fine_tuning.jobs.retrieve("ftjob-abc123")

진행중인 fine-tuning job 취소

client.fine_tuning.jobs.cancel("ftjob-abc123")

fine-tuning model 제거

client.models.delete("ft:gpt-3.5-turbo:acemeco:suffix:abc123")

4. Fine-tuning 모델 사용

fine-tuning job이 끝나면 모델 제작이 완료된 것이니....바로 사용하면 됩니다ㅎ 위에서 설명했다시피 Text-generation API를 사용할 때, 우리는 사용할 모델명을 명시한다. 해당 변수에 우리가 제작한 fine-tuning model 명을 넣어주면 된다.

from openai import OpenAI
client = OpenAI()

completion = client.chat.completions.create(
  model="ft:gpt-3.5-turbo:my-org:custom_suffix:id",
  messages=[
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "Hello!"}
  ]
)
print(completion.choices[0].message)

Parameter

아래의 코드에서와 같이 fine-tuning job을 생성할 때, hyperparameter 설정을 하면 됩니다 :)

client.fine_tuning.jobs.create(
  training_file="file-abc123",
  model="gpt-3.5-turbo",
  hyperparameters={
    "n_epochs":2,
    "batch_size":1,
    "learning_rate_multiplier":1
  }
)

fine-tuning API의 경우 아래 사진에서 보여지는 3가지 parameter에 대해 설명할 것입니다 :)

batch_size

batch_size: 배치 크기는 한 번의 학습(iteration)에 사용되는 데이터 샘플의 수를 의미합니다.

배치 크기가 크면 모델 파라미터가 업데이트되는 빈도가 줄어들지만, 업데이트할 때의 변동성이 낮아집니다. 큰 배치 크기는 계산 효율성을 높일 수 있지만, 때로는 과적합을 피하기 위해 작은 배치 크기가 더 적합할 수 있습니다.

배치 크기에 대한 직관적인 이해를 위해, 훈련 데이터셋에 1000개의 샘플이 있다고 가정해 보겠습니다.

이 경우 배치 크기가 10이라면, 데이터셋은 100개의 배치로 나뉘며, 각 배치에는 10개의 샘플이 포함됩니다. 모델은 각 배치를 처리할 때마다 가중치를 업데이트하고, 이러한 과정을 전체 데이터셋에 대해 반복합니다.

배치 크기가 1이라면 (확률적 경사 하강법), 각 샘플을 독립적으로 처리하고, 각 샘플마다 가중치를 업데이트합니다. 이는 높은 변동성을 가지지만 더 빠른 수렴을 가져올 수 있습니다. 흔히, 배치 사이즈가 작을 때, underfitting(과소적합)의 위험성이 있다고 합니다.

배치 크기가 1000이라면 (전체 배치 학습법), 전체 데이터셋을 한 번에 처리하고, 한 번의 큰 가중치 업데이트를 합니다. 이는 계산적으로 안정적이지만, 메모리 사용이 많고, 때로는 지역 최소값에 갇힐 위험이 있습니다. 흔히, 배치 사이즈가 클 때, overfitting(과적합)의 위험성이 있다고 합니다.

이처럼 배치 크기는 메모리 사용, 학습 속도, 모델의 성능에 영향을 주기 때문에 신중하게 운영할 필요가 있습니다.

learning_rate_multiplier

learning_rate_multiplier: 학습률은 모델이 학습하는 속도나 정도를 조절하는 값입니다.

기본적으로는 모델이 학습할 때 각 파라미터를 얼마나 조정할지 결정합니다.

학습률 배수는 일종의 기본 학습률에 적용되는 스케일링 인자입니다. 이 값을 조정하여 모델의 학습률을 늘리거나 줄일 수 있습니다.

학습률이 너무 높으면 모델이 최적의 해를 지나쳐 버리는 현상이 발생할 수 있어, 결국에는 학습 과정에서 불안정해지고, 모델의 성능이 안좋아집니다.

반면, 너무 낮은 학습률은 모델이 데이터의 핵심 특징을 충분히 학습하기까지 매우 오랜 시간이 걸릴 수 있습니다. 이는 효율적이지 않으며, 오히려 충분한 학습이 이루어지지 않을 경우 예상치 못한 결과를 초래할 수 있습니다.

n_epochs

n_epochs: 전체 훈련 데이터가 학습에 한 번 사용되는 주기를 의미합니다.

즉, 에포크 수는 머신 러닝 모델이 전체 훈련 데이터 세트를 한 번 완전히 통과하는 횟수를 의미합니다.

모델이 한 에포크 동안 훈련 데이터 전체를 한 번 통과하므로, 에포크 수를 늘릴수록 모델은 데이터를 더 많이 볼 수 있으며, 이는 복잡한 문제를 더 잘 학습하는 데 도움이 될 수 있습니다.

하지만 에포크 수가 지나치게 높으면 훈련 시간이 길어지고, 모델이 훈련 데이터에 과도하게 적합되어 과적합을 초래할 수 있습니다.

Tip

토큰 관리는 필수!

기본적으로 토큰은 곧 돈입니다. 따라서, 무작정 사용하다가는 돈쭐날 수 있을 위험이 있으므로 아래 몇 가지 추천드리는 가이드라인을 참고해주세요.

1. max_token 운영으로 사용되는 토큰의 양 제한

max_tokens 변수를 설정하면, 사용할 토큰의 양을 제한할 수 있다. max_token의 값은 입력과 답변의 목적에 따라 달라지겠지만, 이러한 부분도 실험을 통해 조절하길 권장한다.

2. 한글 사용으로 인한 지나친 토큰 소모 지양

기본적으로, 한국어 문자에 사용되는 토큰 개수는 영어 문자에 사용되는 토큰 개수보다 훨씬 많다. 상황에 따라 다르기 때문에, 구체적인 수치를 제시하기 어렵지만 적어도 2배 이상은 소요된다고 예상된다. 굳이 한국어에 대한 입출력이 필요하지 않은 모델이라면, 아래의 사항을 고려해보길 바란다.

• 유료 번역 API 사용: google-translate api, deep-l api (추천), papago api
• 웹 번역기 크롤링: 동적 크롤링을 활용하여, 웹 번역기에 사용할 한국어 텍스트를 입력하고 결과값을 다시 크롤링하여 사용

명확한 실험기준과 평가 파이프라인 구축 중요!

본 글을 요약하자면, OpenAI API를 사용하여 Chat-completion을 하거나 fine-tuning을 할 때 중요한 두가지는 아래와 같다.

• parameter 운영
• prompt-engineering

전자에 대해서는 본 글에서 설명한 개념을 이해하고, 이를 바탕으로 실험기준을 세워야 합니다. 당연히, 파라미터 값은 context와 적용할 도메인의 성격에 따라 달라질 수 있기에 더더욱 명확한 실험기준으로 빠르게 최적해를 찾는 것이 중요합니다.

그러나, 결국 중요한 것은 신뢰성 있는 training_data에 대한 확보 + 도메인 전문가의 평가 라고 생각이 듭니다. 뻔한 소리지만, data의 퀄리티는 늘 중요한 것이겠지요ㅎ

후자에 대해서는, 본 글에서 따로 다루고 있지 않지만 prompt를 잘 구성하는 것도 LLM 성능에 지대한 영향을 끼칩니다. 그러니, 기본적인 prompt 구성법과 이에 대한 실험 기준을 명확하게 하는 것도 중요하다.

최근 mlflow에서는 LLM prompt 운영에 용이한 template도 제공하고 있고, 당연히 모델에 대한 evaluate 파이프라인도 구축할 때 활용할 수 있으므로 mlflow도 적극 활용하길 추천합니다. https://mlflow.org/docs/latest/llms/llm-evaluate/index.html

첨언

• 혹시, 머신러닝에 대한 이해가 부족한 것 같다면 제가 이전에 올린 ML/DL 용어 정리 포스팅을 참고해주시면 좋을 것 같습니다ㅎ
• 꼭, API를 사용하기 전에 OpenAI playground에서 체험해보세요!

참고문헌

• https://platform.openai.com/docs/api-reference
• https://platform.openai.com/playground/chat
• https://www.trustedreviews.com/versus/chat-gpt-4-vs-chat-gpt-3-4309130
• https://www.researchgate.net/figure/Schematic-drawing-of-the-fine-tuned-pipeline-left-branch-and-in-the-zero-shot-pipeline_fig1_374642679
• https://blog.paperspace.com/how-to-maximize-gpu-utilization-by-finding-the-right-batch-size/#resources
• https://etd.ohiolink.edu/apexprod/rws_etd/send_file/send?accession=osu1587693436870594&ref=blog.paperspace.com
• https://www.deeplearningwizard.com/deep_learning/boosting_models_pytorch/lr_scheduling/#step-wise-decay-every-epoch

Airflow Dag를 작성하다 보면 객체 지향 프로그래밍에 대한 중요성을 느낀다. 오랜만에 문법도 익힐 겸 ATM도 구현해보고 이전에 공부했던 전공 책을 꺼내어 OOP 개념과 python class 문법을 복습해보았다.

python class 구현 방법

파이썬에서 클래스를 구현하는 기본 방법을 간단한 계산기 구현 예제로 살펴보겠습니다.

 class Calculator:
     def __init__(self):
         self.result = 0

     def add(self, num):
         self.result += num
         return self.result

 cal1 = Calculator()
 cal2 = Calculator()

 class FourCal:
     def __init__(self, first, second):
         self.first = first
         self.second = second
     def setdata(self, first, second):
         self.first = first
         self.second = second
     def add(self):
         result = self.first + self.second
         return result

 class MoreFourCal(FourCal):
     pass

 a = MoreFourCal(4,2)
 print(a.add())

이 예제에서는 기본적인 클래스 구현과 상속에 대해 보여줍니다. Calculator 클래스는 기본적인 덧셈 기능을 제공하며, FourCal 클래스는 두 숫자를 더하는 메서드를 포함합니다. MoreFourCal 클래스는 FourCal 클래스를 상속받아 추가 기능 없이 그대로 사용합니다. 이제 이를 기반으로 ATM기를 구현하여 객체 지향 프로그래밍(OOP)의 원칙을 다시 익혀보겠습니다.

ATM기를 구현하며 OOP를 다시 익히다

OOP를 처음 접하는 분들에게는 어떠한 것도 참고하지 마시고 최소 기능만 구상하시고 ATM 구현해보시는 것을 추천합니다.

ATM, ATM controller Class 구현

# Import Bank class from the bank module

from bank import Bank

# Define the class ATM which represents the ATM machine

class ATM:

    # Constructor for ATM class. It takes a Bank object as argument

    def __init__(self, bank: Bank):
        self.bank = bank
        self.card_number = None
        self.account_number = None

    # Method for inserting card. It validates the card_number with the bank

    def insert_card(self, card_number: str) -> None:
        if self.bank.validate_card(card_number):
            self.card_number = card_number
        else:
            raise ValueError("Invalid card")

    # Method for entering PIN. It validates the PIN with the bank

    def enter_pin(self, pin: str) -> None:
        if not self.bank.validate_pin(self.card_number, pin):
            raise ValueError("Invalid PIN")

    # Method for getting a list of accounts linked to the card

    def get_accounts_list(self) -> list:
        return self.bank.get_accounts(self.card_number)

    # Method for selecting an account. It validates if the account_name exists in the bank

    def select_account(self, account_name: str) -> None:
        if account_name not in self.bank.get_accounts(self.card_number):
            raise ValueError("Account does not exist")
        self.account_number = account_name

    # Method for checking the balance of the selected account

    def check_balance(self) -> int:
        return self.bank.get_balance(self.card_number, self.account_number)

    # Method for depositing money into the selected account

    def deposit_money(self, amount: int) -> None:
        self.bank.deposit(self.card_number, self.account_number, amount)

    # Method for withdrawing money from the selected account

    def withdraw_money(self, amount: int) -> None:
        self.bank.withdraw(self.card_number, self.account_number, amount)

# Define the class ATMController which controls the ATM machine

class ATMController:
    def __init__(self, atm: ATM):
        self.atm = atm

    def insert_card(self, card_number: str) -> str:
        try:
            self.atm.insert_card(card_number)
            return "Your card has been successfully inserted"
        except ValueError as e:
            return str(e)

    def enter_pin(self, pin: str) -> str:
        try:
            self.atm.enter_pin(pin)
            return "PIN number entered successfully"
        except ValueError as e:
            return str(e)

    def get_accounts_list(self) -> list:
        try:
            return self.atm.get_accounts_list()
        except ValueError as e:
            return str(e)

    def select_account(self, account_name: str) -> str:
        try:
            self.atm.select_account(account_name)
            return "Account has been successfully selected"
        except ValueError as e:
            return str(e)

    def check_balance(self) -> int:
        try:
            return self.atm.check_balance()
        except ValueError as e:
            return str(e)

    def deposit_money(self, amount: int) -> str:
        try:
            self.atm.deposit_money(amount)
            return "Deposit has been processed"
        except ValueError as e:
            return str(e)

    def withdraw_money(self, amount: int) -> str:
        try:
            self.atm.withdraw_money(amount)
            return "Withdrawal has been processed"
        except ValueError as e:
            return str(e)

ATM Class workflow

• ATM 클래스는 Bank 클래스의 객체를 받아 초기화됩니다.
• insert_card, enter_pin, get_accounts_list, select_account, check_balance, deposit_money, withdraw_money 메서드를 통해 ATM의 기본 기능을 수행합니다.
• 각 메서드는 Bank 클래스의 메서드를 호출하여 필요한 작업을 수행합니다.

ATM Controller Class workflow

• ATMController 클래스는 ATM 객체를 받아 초기화됩니다.

• 각 메서드는 ATM 클래스의 메서드를 호출하여 결과를 반환하거나 예외를 처리합니다.
• insert_card, enter_pin, get_accounts_list, select_account, check_balance, deposit_money, withdraw_money 메서드를 통해 ATM의 기능을 제어합니다.

Bank Class 구현

from abc import ABC, abstractmethod

# This class represents the blueprint for any bank

class Bank(ABC):

    # Method to check if a card is valid

    @abstractmethod
    def validate_card(self, card_number: str) -> bool:
        pass

    # Method to verify the entered PIN

    @abstractmethod
    def validate_pin(self, card_number: str, pin: str) -> bool:
        pass

    # Method to retrieve all accounts associated with a card

    @abstractmethod
    def get_accounts(self, card_number: str) -> list:
        pass

    # Method to check balance of an account

    @abstractmethod
    def get_balance(self, card_number: str, account_number: str) -> int:
        pass

    # Method to deposit money into an account

    @abstractmethod
    def deposit(self, card_number, account_number: str, amount: int) -> None:
        pass

    # Method to withdraw money from an account

    @abstractmethod
    def withdraw(self, card_number, account_number: str, amount: int) -> None:
        pass

# MockBank is a basic bank for testing purposes

class MockBank(Bank):

    # MockBank has some predefined accounts for testing

    def __init__(self):
        self.accounts = {
            "123456": {"pin": 1234, "accounts": {"account1": 5000, "account2": 10000}},
            "654321": {"pin": 4321, "accounts": {"account1": 3000, "account2": 20000}}
        }

    def validate_card(self, card_number: str) -> bool:
        return card_number in self.accounts

    def validate_pin(self, card_number: str, pin: int) -> bool:
        return self.accounts[card_number]["pin"] == pin

    def get_accounts(self, card_number: str) -> list:
        return list(self.accounts[card_number]["accounts"].keys())

    def get_balance(self, card_number: str, account_name: str) -> int:
        return self.accounts[card_number]["accounts"][account_name]

    def deposit(self, card_number: str, account_name: str, amount: int) -> None:
        self.accounts[card_number]["accounts"][account_name] += amount

    def withdraw(self, card_number: str, account_name: str, amount: int) -> None:
        if self.accounts[card_number]["accounts"][account_name] < amount:
            raise ValueError("balance is insufficient")
        self.accounts[card_number]["accounts"][account_name] -= amount

Bank Class workflow

• Bank 클래스는 추상 클래스(ABC)로 정의되어 있으며, 여러 추상 메서드를 포함합니다.

• MockBank 클래스는 Bank 클래스를 상속받아 실제 구현을 제공합니다.
• validate_card, validate_pin, get_accounts, get_balance, deposit, withdraw 메서드를 통해 카드 유효성 검사, PIN 검사, 계좌 조회, 잔액 조회, 입금 및 출금 기능을 제공합니다.

컨트롤러 테스트 by unittest

from bank import MockBank
from atm import ATM, ATMController
import unittest

class ATMControllerTest(unittest.TestCase):
    def setUp(self):
        self.bank = MockBank()
        self.atm = ATM(self.bank)
        self.atm_controller = ATMController(self.atm)

    def test_insert_card(self):
        result = self.atm_controller.insert_card("123456")
        self.assertEqual(result, "Your card has been successfully inserted")

    def test_enter_pin(self):
        self.atm_controller.insert_card("123456")
        result = self.atm_controller.enter_pin(1234)
        self.assertEqual(result, "PIN number entered successfully")

    def test_get_accounts_list(self):
        self.atm_controller.insert_card("123456")
        result = self.atm_controller.get_accounts_list()
        self.assertEqual(result, ["account1", "account2"])

    def test_select_account(self):
        self.atm_controller.insert_card("123456")
        self.atm_controller.get_accounts_list()
        result = self.atm_controller.select_account("account1")
        self.assertEqual(result, "Account has been successfully selected")

    def test_check_balance(self):
        self.atm_controller.insert_card("123456")
        self.atm_controller.get_accounts_list()
        self.atm_controller.select_account("account1")
        result = self.atm_controller.check_balance()
        self.assertEqual(result, 5000)

    def test_deposit_money(self):
        self.atm_controller.insert_card("123456")
        self.atm_controller.get_accounts_list()
        self.atm_controller.select_account("account1")
        result = self.atm_controller.deposit_money(1000)
        self.assertEqual(result, "Deposit has been processed")

    def test_withdraw_money(self):
        self.atm_controller.insert_card("123456")
        self.atm_controller.get_accounts_list()
        self.atm_controller.select_account("account1")
        result = self.atm_controller.withdraw_money(3000)
        self.assertEqual(result, "Withdrawal has been processed")

if __name__ == '__main__':
    unittest.main()

# if hard to understand unittest, use this
# def test_ATMController():

#     bank = MockBank()
#     atm = ATM(bank)
#     atm_controller = ATMController(atm)

#     result = atm_controller.insert_card("654321")
#     assert result == "Your card has been successfully inserted"

#     result = atm_controller.enter_pin(4321)
#     assert result == "PIN number entered successfully"

#     result = atm_controller.get_accounts_list()
#     assert result == ["account1", "account2"]

#     result = atm_controller.select_account("account1")
#     assert result == "Account has been successfully selected"

#     result = atm_controller.check_balance()
#     assert result == 3000

#     result = atm_controller.deposit_money(1000)
#     assert result == "Deposit has been processed"

#     result = atm_controller.withdraw_money(3000)
#     assert result == "Withdrawal has been processed"

# test_ATMController()

Test Class workflow

• ATMControllerTest 클래스는 unittest.TestCase를 상속받아 ATM의 다양한 기능을 테스트합니다.

• setUp 메서드를 통해 테스트 환경을 초기화합니다.
• 각 테스트 메서드는 ATMController 클래스의 메서드를 호출하여 기능을 검증합니다.

해당 예제를 통해 기본적인 클래스 구현, 상속, 추상 클래스 사용 및 유닛 테스트 작성 방법을 알아보았습니다.

Airflow DAG 작성에서 객체 지향 프로그래밍(OOP)의 중요성

개인적인 생각으로, Airflow DAG(DAG: Directed Acyclic Graph)을 작성할 때 객체 지향 프로그래밍(OOP)이 중요한 이유는 특정 작업(task)을 클래스로 정의하고, 이를 여러 DAG에서 재사용하는 것이 Airflow 아키텍처의 지향점에 부합하다고 생각해서이다.

코드 예시: OOP를 활용한 Airflow DAG 작성

Step 1: 공통 작업 클래스를 정의

from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults

class MyTaskOperator(BaseOperator):
    @apply_defaults
    def __init__(self, param1, param2, *args, **kwargs):
        super(MyTaskOperator, self).__init__(*args, **kwargs)
        self.param1 = param1
        self.param2 = param2

    def execute(self, context):
        # 작업 실행 로직
        print(f"Executing task with {self.param1} and {self.param2}")

Step 2: DAG 작성 시 작업 클래스를 활용

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from datetime import datetime

# DAG 기본 설정
default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1),
    'retries': 1,
}

# DAG 정의
with DAG(dag_id='my_dag', default_args=default_args, schedule_interval='@daily') as dag:
    start = DummyOperator(task_id='start')

    # MyTaskOperator를 사용하여 작업 생성
    task1 = MyTaskOperator(task_id='task1', param1='value1', param2='value2')
    task2 = MyTaskOperator(task_id='task2', param1='value3', param2='value4')

    end = DummyOperator(task_id='end')

    # 작업 의존성 설정
    start >> task1 >> task2 >> end

코드 설명

1. MyTaskOperator 클래스 정의:

   • MyTaskOperator는 BaseOperator를 상속받아 구현한 클래스입니다.
   • __init__ 메서드에서 필요한 매개변수를 초기화합니다.
   • execute 메서드에서 실제 작업을 수행하는 로직을 작성합니다.

2. DAG 정의:

   • DAG를 정의할 때 MyTaskOperator를 사용하여 작업을 생성합니다.
   • start, task1, task2, end 작업을 정의하고, 작업 간의 의존성을 설정합니다.

결론

Airflow DAG를 작성할 때 객체 지향 프로그래밍(OOP)을 활용하면 코드의 재사용성, 유지보수성, 확장성, 가독성을 크게 향상시킬 수 있기에, 예시처럼 작업을 클래스로 정의하고 이를 여러 DAG에서 재사용하면, 복잡한 워크플로우를 보다 효율적으로 관리할 수 있습니다 :)

ATM 코드에 대한 실행방법과 추가로 현금함을 구현하기 위한 가이드라인은 아래 github 링크의 README.md에 포함되어 있습니다 :) 소스코드: https://github.com/idle-danie/OOP_atm

많이 사용되지만 혼동되는 ML/DL 관련 용어들과 개념들에 대해 정리해보았습니다 :)

AI vs ML vs DL

Components of Deep Learning

• Data: 이 문제를 풀기 위해 어떤 데이터를 사용해야할까?

• Model: 어떤 모델이 원하는 결과를 최대한 잘 도출할까? ex) GPT-3, LSTM, GAN, ResNet

• Loss function : 모델을 어떻게 학습시킬까? ex) Regression type -> MAE, MSE, RMSE Classification type -> Binary cross-entropy, Categorial cross-entropy

• Optimization Algorithm: 네트워크를 어떻게 줄일까?

Overview of Deep Learning history posted by Denny Britz

시간에 따른 딥러닝 모델의 발전과정 overview에 대한 글인데, 흥미롭게 읽어서 가져와보았습니다 :) https://dennybritz.com/posts/deep-learning-ideas-that-stood-the-test-of-time/

• AlexNet (2012)

• DQN (2013)

• Seq2Seq by Attention (2014)

• Adam Optimizer (2014)

• GAN (2014, 2015)

• ResNet (2015)

• Transformer (2017)

• BERT and fine tuned models (2018)

• Large Language models like GPT-3 (2019~)

Optimization

• Generalization: 학습된 모델이 unseen data에서도 work well?

  • Generalization performance => Generalization gap = (Test error - Training error)

• Overfitting: Training data에서는 well work, Test data에서는 not well work

• Underfitting: 네트워크가 간단하거나 train이 부족해서 Training data에서도 not well work

• parameter: 최적해에서 찾고 싶은 값 (ex: weight, bias)

• hyperparameter: output을 결정하는 변수 (ex: learning rate; 어떤 loss function을 사용할 것인지?)

• Cross-validation : Training data를 partition하여 Train data, Validation data에 적용 (Training data -> Training data + Validation data)

  • 최적의 hyperparameter set을 찾고 고정한 상태에서 학습시킬 때는 모든 데이터 활용 (test data x)

• Bias: 얼마나 목표 타겟에 가깝나

  • low bias: 타겟에 가깝다
  • high bias: 타겟에서 멀다

• Variance: 얼마나 모여있는지

  • low variance -> 잘 모여있다
  • high variance : overfitting 가능성 큼

• Bias and Variance Tradeoff : bias와 variance를 동시에 줄이기는 쉽지 않다

• Bootstrapping: dataset에서 무작위로 표본을 추출하여 여러 예측 모델 생성 (any test or metric that uses random sampling)

• Bagging (Boostrapping aggregating) 앙상블 : 독립적으로 고정된 학습데이터로 모델 여러개를 훈련 (averaging or voting)

  • 예) 10만개의 학습데이터로 하나의 모델을 학습하지 않고, 80%로 n개의 모델을 돌리고 값의 평균 또는 voting 출력값을 사용

참고문헌

1. https://sungwookkang.com/1409
2. https://brunch.co.kr/@mnc/9
3. https://dennybritz.com/posts/deep-learning-ideas-that-stood-the-test-of-time/

본 글은 Baekjoon에서 제공하는 알고리즘 문제 및 유저 데이터를 분석하는 [solved.ac] (https://solved.ac/) 서비스 및 API를 사용하여 관련 데이터를 추출한 뒤, DW/DM을 생성하고 ETL을 구축하는 프로젝트의 일부 과정에 해당하는 내용입니다 :)

본 내용에서 설명하는 구체적인 상황과 문제점은 아래와 같습니다.

상황

solved.ac API를 호출하여, 알고리즘 문제 관련 데이터를 추출하여 csv에 저장해야 한다.

아무런 성능 개선 코드가 적용되지 않은 기본 API호출 코드는 아래와 같을 것이다.

import requests

url = "https://~~~"
headers = {"Accept": "application/json"}

page = 1
while True:
    querystring = {"query": '', "page": str(page)}
    response = requests.get(url, headers=headers, params=querystring)
    data = response.json()

    for item in data['items']:
        if item['tier'] < 27:  # If the 'tier' is less than 7
            break
    page += 1

문제점

일단 프로젝트 여건 상 로컬에서 파이프라인 실행해야 하기 때문에, 데이터가 커질수록, 메모리나 디스크 i/o와 같은 성능 이슈 발생한다. 또한, 위와 같은 이유일 수도 있겠지만 기본적으로 API를 호출하여 데이터를 추출하여 DW/DM에 적재하는 총 파이프라인 실행시간이 매우 길다.

문제가 발생한 이유를 pandas를 이용하여 데이터를 csv에 저장하는 방식에서 병목이 일어났다고 판단하여 디스크 i/o와 Memory를 활용하여 csv에 데이터를 저장하는 방식을 비교해보려고 한다. 또한, 전체적인 파이프라인 실행속도를 높이기 위해 우선 데이터를 추출하는 과정에서 파이썬 비동기 (async)를 도입한다.

디스크 i/o vs Memory

일반적인 상식으로 생각해보면, 메모리를 활용한 방식이 더 빠를 것이고 코드의 가독성도 좋다.

아무리 좋은 SSD를 구비하여 i/o속도를 높여도, 메모리 용량을 업그레이드하는 것보다 컴퓨터 성능 개선에 효과적이지 않다는 것을 들어 보았을 것이다.

pandas 공식 문서에서도 메모리를 활용한 방식을 채택한다. https://pandas.pydata.org/docs/user_guide/scale.html

그래도 일단 비교를 위해 두 방식을 모두 실행해보자.

디스크 i/o를 활용한 추출

한번에 모든 데이터를 csv를 적재하는 것이 아닌 일정 bunch단위를 미리 정해놓고, 데이터가 정해놓은 데이터 단위 기준에 다다르면 그때마다 csv를 적재하는 방식이다.

디스크 i/o를 활용한 code

async def collect_data_and_save_to_csv():
    url = "https://~~~"

    async with aiohttp.ClientSession() as session:
        count = await fetch_page(session, url, 0, ua.random)
        max_page = count['count'] // 50 + 1
        csv_file = os.path.join(os.getcwd(), 'async_get_user.csv')
        start_time = time.time()
        user_agent = ua.random

        try:
            page = 1
            while page <= max_page:
                if page % 100 == 0:
                    user_agent = ua.random

                data = await fetch_page(session, url, page, user_agent)
                if data is None:
                    continue
                items = data['items']

                filtered_items = []
                for item in items:
                    if item['tier'] == 6:
                        print("Tier 6 reached. Stopping data collection.")
                        return

                    filtered_item = {
                        'handle': item['handle'],
                        'solvedCount': item['solvedCount'],
                        'tier': item['tier'],
                        'rating': item['rating'],
                        'ratingByProblemsSum': item['ratingByProblemsSum'],
                        'ratingByClass': item['ratingByClass'],
                        'ratingBySolvedCount': item['ratingBySolvedCount'],
                        'ratingByVoteCount': item['ratingByVoteCount'],
                        'class': item['class'],
                        'maxStreak': item['maxStreak'],
                        'joinedAt': item['joinedAt'],
                        'rank': item['rank']
                    }
                    filtered_items.append(filtered_item)

                df = pd.DataFrame(filtered_items)
                if page == 1:
                    df.to_csv(csv_file, index=False, encoding='utf-8')
                else:
                    df.to_csv(csv_file, mode='a', header=False, index=False, encoding='utf-8')

                page += 1
                print(f"현재 페이지: {page}")

        except Exception as e:
            print(f"An error occurred: {e}")
            return

        end_time = time.time()
        execution_time = end_time - start_time
        print(f"Task 실행 시간: {execution_time}초")
        print("Data collection and saving completed successfully.")

실행시간

Tier 6 reached. Stopping data collection. 총 소요 시간: 512.5091943740845

메모리를 활용한 추출

위와 달리, 데이터를 호출하고 메모리에 담은 뒤 한번에 csv형태로 저장하는 방식이다.

async def collect_data():
    url = "https://~~~"

    async with aiohttp.ClientSession() as session:
        count = await fetch_page(session, url, 0, ua.random)
        max_page = count['count']//50+1
        data_list = []

        try:
            page = 1
            user_agent = ua.random
            while page <= max_page:
                if page % 100 == 0:
                    user_agent = ua.random

                data = await fetch_page(session, url, page, user_agent)
                if data is None:
                    continue
                items = data['items']

                filtered_items = []
                for item in items:
                    if item['tier'] == 6:
                        print("Tier 6 reached. Stopping data collection.")
                        return data_list

                    filtered_item = {
                        'handle': item['handle'],
                        'solvedCount': item['solvedCount'],
                        'tier': item['tier'],
                        'rating': item['rating'],
                        'ratingByProblemsSum': item['ratingByProblemsSum'],
                        'ratingByClass': item['ratingByClass'],
                        'ratingBySolvedCount': item['ratingBySolvedCount'],
                        'ratingByVoteCount': item['ratingByVoteCount'],
                        'class': item['class'],
                        'maxStreak': item['maxStreak'],
                        'joinedAt': item['joinedAt'],
                        'rank': item['rank']
                    }
                    filtered_items.append(filtered_item)

                data_list.extend(filtered_items)
                print(f"현재 페이지: {page}")
                page += 1

        except Exception as e:
            print(f"오류 발생: {e}")
            return

        return data_list

실행시간

아래와 같이 특정 상황에서 멈추는 상황이 발생하였다. 여기서 주목할 점은, 멈추는 페이지의 지점이 실행 시마다 비슷하다는 점이다. 따라서, 디스크 i/o를 활용하는 방식을 채택하지 않는다면, 메모리 제한 및 gc 관리가 필요할 것으로 예측할 수 있다.

Problem

사실 현재 측정하는 실행 속도는 로컬 환경의 요소로만 결정되지는 않는다. 외부적 요소인 네트워크 상태와 request수 제한, 그리고 내부적 요소인 python garbage collection와 pandas 메모리 할당 등의 여러 변수가 있기 때문에 총 소요시간을 통해 명확한 결론을 낼 수는 없다. 단지, 이러한 모든 가능성을 인지하고 상황에 따른 엔지니어링을 해야한다는 것이 중요하다.

일단, 문제점을 정리해보면 아래와 같을 것이다.

외부적 요인: 네트워크, API request 제한

주의 크롤링을 할 때는 robots.txt, API를 호출할 때는 어디가에 숨어있을 API가이드라인을 통해 명확히 request수 제한과 같은 제한사항을 꼭 체크하자. 이미 IP block이 이루어졌다면, 제한이 풀리길 기다리거나 VPN을 사용하는 방법을 사용해야 하니, 미리미리 인지하고 따르자.

어떠한 내부적인 요소를 fix하여 퍼포먼스를 개선한다 해도, 네트워크 환경이 작업을 제한하면 의미가 없다.

제한을 지키며 극한의 퍼포먼스를 내기 위해서는 최대한 API 가이드라인에 따라 호출 시간을 sleep()을 활용하여 지연시키거나, asyncio에서 await을 사용하여 API 서버에 피해를 주지 않아야 한다.

위 사항으로 만족(?)되지 않는다면, 극한의 최적화를 위한 다음과 같은 방법들이 있을 수 있다. (권장 X)

1. User agent 변경 Python에서는 fake user agent 라이브러리를 활용하여 매 호출, 혹은 특정 횟수마다 user-agent를 변경해주면 다른 유저로 인식하여 request수 제한 같은 요소를 피할 수 있다. 하지만, 이러한 이상한(?)행위도 봇으로 쉽게 식별될 수 있다는 점을 인지하자.
2. semaphore 활용 멀테쓰레딩을 사용하고 있다면, semaphore로 쓰레드 수를 제한하여 실행 속도를 조절할 수 있다.

내부적 요인: 메모리 부하

한번에 모든 데이터를 메모리에 저장하고, 이를 변환하는 행위는 메모리의 부하를 발생시킬 수 있다. 해결책은 뻔하겠지만, 나눠서 변환하면 된다! 또한, gc를 사용하여 적절히 메모리 관리를 개선할 수 있다. 아래 코드는 10페이지마다 한 번씩 gc.collect()를 호출하여 메모리 관리를 개선한 code이다.

import gc
import aiohttp
import asyncio
from fake_useragent import UserAgent

ua = UserAgent()

async def fetch_page(session, url, page, user_agent):
    headers = {'User-Agent': user_agent}
    params = {'page': page}
    async with session.get(url, headers=headers, params=params) as response:
        if response.status == 200:
            return await response.json()
        return None

async def collect_data():
    url = "https://~~~"

    async with aiohttp.ClientSession() as session:
        count = await fetch_page(session, url, 0, ua.random)
        max_page = count['count'] // 50 + 1
        data_list = []

        try:
            page = 1
            user_agent = ua.random
            while page <= max_page:
                if page % 100 == 0:
                    user_agent = ua.random

                data = await fetch_page(session, url, page, user_agent)
                if data is None:
                    page += 1
                    continue
                items = data['items']

                filtered_items = []
                for item in items:
                    if item['tier'] == 6:
                        print("Tier 6 reached. Stopping data collection.")
                        return data_list

                    filtered_item = {
                        'handle': item['handle'],
                        'solvedCount': item['solvedCount'],
                        'tier': item['tier'],
                        'rating': item['rating'],
                        'ratingByProblemsSum': item['ratingByProblemsSum'],
                        'ratingByClass': item['ratingByClass'],
                        'ratingBySolvedCount': item['ratingBySolvedCount'],
                        'ratingByVoteCount': item['ratingByVoteCount'],
                        'class': item['class'],
                        'maxStreak': item['maxStreak'],
                        'joinedAt': item['joinedAt'],
                        'rank': item['rank']
                    }
                    filtered_items.append(filtered_item)

                data_list.extend(filtered_items)
                print(f"현재 페이지: {page}")
                page += 1

                # 특정 간격으로 가비지 컬렉션 실행
                if page % 10 == 0:
                    gc.collect()

        except Exception as e:
            print(f"오류 발생: {e}")
            return

        return data_list

위와 같이 코드 개선은 기존에 비해 20배 이상의 실행속도 개선을 가져왔다.

실행시간 (개선)

Tier 6 reached. Stopping data collection. 총 소요 시간: 25.1141256242419

결국은 Multi-Process?

모든 개선 과정을 끝내고 문득 이러한 생각이 들었다. 그냥 멀티 프로세싱을 쓰면 안되나? 이러한 흐름으로 처음 멀티 프로세싱을 접하게 되었다.

일반적으로, 실행속도 개선을 위해서 멀티 쓰레드, 비동기 호출, 멀티 프로세싱을 활용할 수 있다. 각각의 대한 정의를 간단하게 정리해보면 아래와 같을 것이다.

*멀티프로세싱 (Multi-processing) * 멀티프로세싱은 여러 프로세스를 생성하여 병렬로 작업을 수행하는 방식으로, 각 프로세스는 별도의 메모리 공간을 사용하여 CPU 집약적인 작업에서 성능을 향상시킨다.

*비동기 (Asynchronous) * 비동기 프로그래밍은 작업이 완료될 때까지 기다리지 않고 다른 작업을 계속 수행하는 방식으로, 주로 io 바운드 작업에서 효율성을 높이고 응답성을 개선한다.

*멀티쓰레딩 (Multit-hreading) * 멀티쓰레딩은 단일 프로세스 내에서 여러 쓰레드를 생성하여 병렬로 작업을 수행하는 방식으로, 메모리를 공유하며 CPU 및 io 바운드 작업의 성능을 향상시킨다.

하지만 python은 GIL때문에 멀티 쓰레딩에 제한적일 수 밖에 없고 데이터의 매우 커질 경우 극한의 퍼포먼스를 위해서는 병렬처리를 선택할 수 밖에 없다고 생각한다. 멀티 프로세싱을 적용하여 최종 코드를 개선하면 아래와 같을 것이다.

import gc
import aiohttp
import asyncio
from fake_useragent import UserAgent
from concurrent.futures import ProcessPoolExecutor

ua = UserAgent()

async def fetch_page(session, url, page, user_agent):
    headers = {'User-Agent': user_agent}
    params = {'page': page}
    async with session.get(url, headers=headers, params=params) as response:
        if response.status == 200:
            return await response.json()
        return None

async def process_page(session, url, page):
    user_agent = ua.random
    data = await fetch_page(session, url, page, user_agent)
    if data is None:
        return []

    items = data['items']
    filtered_items = []
    for item in items:
        if item['tier'] == 6:
            print("Tier 6 reached. Stopping data collection.")
            return filtered_items

        filtered_item = {
            'handle': item['handle'],
            'solvedCount': item['solvedCount'],
            'tier': item['tier'],
            'rating': item['rating'],
            'ratingByProblemsSum': item['ratingByProblemsSum'],
            'ratingByClass': item['ratingByClass'],
            'ratingBySolvedCount': item['ratingBySolvedCount'],
            'ratingByVoteCount': item['ratingByVoteCount'],
            'class': item['class'],
            'maxStreak': item['maxStreak'],
            'joinedAt': item['joinedAt'],
            'rank': item['rank']
        }
        filtered_items.append(filtered_item)

    return filtered_items

async def collect_data():
    url = "https://~~~"

    async with aiohttp.ClientSession() as session:
        count = await fetch_page(session, url, 0, ua.random)
        max_page = count['count'] // 50 + 1

        loop = asyncio.get_event_loop()
        tasks = []
        with ProcessPoolExecutor() as executor:
            for page in range(1, max_page + 1):
                tasks.append(loop.run_in_executor(executor, process_page, session, url, page))

            data_list = await asyncio.gather(*tasks)

        # Flatten the list of lists
        data_list = [item for sublist in data_list for item in sublist]

        # 특정 간격으로 가비지 컬렉션 실행
        gc.collect()

        return data_list

하지만, 로컬과 같이 제한된 상황에서 core 수를 늘릴 수 있는 상황이 아니면, 사실상 멀티프로세싱을 통한 개선은 의미가 없다.

결국, 앞서 i/o vs Memory에서의 논점과 같이 문제가 발생하는 상황에 따라 적절한 대응방식을 상이하기 때문에, 단순 비교를 하는 것은 불가능하다.

고찰

결국 개발자의 실력은 기술적 이해도를 바탕으로 상황에 맞는 최적의 방법을 도출하는 수준에 비례하는 것 같다. 그래서, 경험과 지식을 기반으로 나름의 기준을 미리 설정하는 습관이 중요하다고 생각한다.

나는 고민의 시간을 줄이기 위해 나름의 3가지 고려사항 및 기준을 세웠다.

1. 간단한 태스크인가? 개인적으로 가지고 있는 기준은 일단 간단한 태스크라면 15분 안에 실행이 끝나야 한다. 15분의 기준은 AWS Lambda와 같은 severless 환경에서 요구하는 최대 실행 시간이다. 말 그대로, severless에서 task를 돌릴만큼의 심플한 태스크이어야 한다. (Airflow의 과부하를 줄이고 간단한 task는 Lambda에서 진행 -> Lambda task call Dag)
2. 쓰레드 간의 연관성이 결과에 주요한 영향을 미치지 않아야 한다. 예를 들어, 크롤링 작업이라면 페이지 순서가 고려하지 않아도 된다는 점이다.
3. 하나의 쓰레드가 실패한다면? 하나의 쓰레드가 실패하면 아래의 사진처럼 전체 프로세스에 영향을 미치게 된다. 따라서 하나의 쓰레드가 하는 일을 정확하게 파악하는 것이 중요하다.

참고문헌

https://eunjinii.tistory.com/41

ps. 2024년에서 이 글을 다시 보니, 약간 민망하다. 하지만, 당시에는 해당 인사이트를 얻기까지 엄청난 고뇌의 시간이 들었던 것으로 기억한다.

데이터를 통해 사용자의 서비스 경험을 개선하고 운영비용을 감소시켜 결과적으로 회사의 의사결정을 도울 수 있다. 하지만 전적으로 데이터만 믿고 따라가기만 한다고 성공을 보장할 수 없다. 따라서, 아래의 data decision을 대표하는 두 정의를 살펴보고 '데이터에 의한 결정은 무조건 옳다'라는 편협한 사고를 없애야 한다. 결과적으로, 회사의 크기와 상황에 맞는 결정론을 기반으로 의사결정을 진행해야 한다는 것이 중요하다.

Data driven decision vs Data informed decision

Data driven decision : 결정의 주체와 수단이 모두 데이터인 data-decison

기본적으로 데이터를 기반으로 의사결정을 진행합니다. 개인의 주관을 배제하고, 어떤 선택이 나은지 결정할 수 있는 가설을 테스트 및 검증하여 결과를 의사결정에 반영하는 것입니다.

Data informed decision : 위와 같이 결정의 주체는 사람이지만, 수단에는 데이터를 포함한 다른 결정 요소들이 개입하는 data-decison

데이터를 보고 주관적인 의견을 생성하여 결론을 도출하는 것이라고 이해하면 될 것 같다. 따라서, 해당 결정을 진행하기 위해서는 기본적으로 Data Team의 인원들과 의사결정 주체가 주요 지표와 도메인 지식에 대한 이해도가 필수적으로 요구될 것이다.

Data Team 구성

모든 데이터 팀이 아래와 같이 이루어져 있다고 말할 수 없지만, 성숙한 데이터 팀의 경우 보통 아래 사진과 같은 구성을 보인다. (그림에 재능은 없으니 이해해 주시길...ㅎ) 보통 Data Team의 구성원들은 아래와 같은 직군으로 분류된다. 🔼 = Data Scientist ⏹️ = Data Analyst ⏺️ = Data Engineer

데이터 분석가

• 데이터 웨어하우스의 데이터를 기반으로 지표를 만들고 대시보드를 통한 시각화
• 내부 직원들의 데이터 관련 질문 응답

데이터 사이언티스트

• 인공지능 모델을 개발하여 서비스 개선 (개인화, 자동화, 최적화)

데이터 엔지니어는 아래에서 따로 설명하려고 한다 :)

Data Engineer's Role

소프트웨어 엔지니어링

• 기본적으로 벡엔드 프레임워크를 사용하여 API를 개발할 수 있어야 한다고 생각한다. 여러가지 상황이 있겠지만 데이터를 처리하는 전반적인 과정일 수 있고, 머신러닝 엔지니어의 역할을 한다면 프로덕션 단계에서 ML/DL 모델이 API로써 역할을 할 수 있게 만들어 줘야 할 것이다. 또한 데이터 분석가에게도 필요한 API나 데이터를 공급해야 한다.

데이터 웨어하우스 (DW) 구축

• 주로 클라우드 서비스로 관리하는 추세이다 (ex: BigQuery, Redshift, Snowflake)

데이터 파이프라인 구축 (ETL)

• ETL: 기존의 데이터베이스에서 원하는 정보를 데이터 웨어하우스에 적재하는 것 ⇒ 추출(Extract), 변환(Transform), 로드(Load)의 과정을 의미한다. 이러한 과정에 필요한 스케줄링은 보통 Airflow로 많이 관리한다.

• ELT: 데이터 웨어하우스에 적재된 데이터를 새로운 정보 혹은 요약된 정보로 제공하는 것이다. 주로, DBT가 해당 역할을 포함하여 늘 데이터 엔지니어를 피곤한게 하는 data transform 과정을 효율적으로 처리하게 해주고 있어, 많은 회사들이 해당 기술을 채용하고 있다.

• 어느정도 성숙한 Data team에서는 Amazon S3와 같은 서비스를 이용하여 데이터 레이크를 구축하는데, 위 사진의 초록 화살표의 과정에서는 Spark와 같은 빅데이터 처리 시스템이 많이 쓰인다.

Why OLAP?

아예 데이터 엔지니어링 분야를 모르거나, 개발을 처음 접하는 분들은 문득 이런 생각이 들 수 있다. _'그냥 Backend Engineer가 서비스 DB에 적재된 데이터를 기반으로 위의 역할들을 잘 소화하면 되는 것 아닌가?' _

아마, 이러한 물음은 '서비스를 하고 있는 OLTP 시스템 내에서 그냥 분석하면 되는 것 아니야?' 라는 물음으로 치환되어 발생하는 것으로 생각된다. 하지만, OLTP (Online Transaction Processing) 에서 직접 분석을 실행하기에는 한계점이 매우 많다.

물론, 일회성 분석이나 소규모 기업에서는 OLTP에서 직접 분석을 실행하는 경우도 있지만, 단기적으로는 효과적일 수 있어도 궁극적으로 확장성이 매우 떨어진다. 어떤 시점에서 OLTP의 구조적 제한이나 경쟁 트랜잭션 워크로드와의 리소스 경합 때문에 성능 문제가 발생할 수 있다.

따라서, 트랜잭션을 처리하기 위해 설계 된 시스템이 아닌, 대규모 분석 쿼리를 실행하도록 구축되어 있는 OLAP (Online Analytical Processing) 시스템이 필요하다.

보통의 OLAP 시스템은 열 기반으로 설계되어 있어, 대량의 데이터를 스캔하도록 최적화 되어 있습니다. 우리가 위에서 언급한 DW 솔루션들은 모두 같은 특징을 가지고 있다. 참고로, OLAP 시스템에서는 인덱스나 PK와 같은 OLTP에서 우리가 빠르게 데이터를 조회하기 위해서 사용하는 것들은 사용되지 않는다. 모든 쿼리에는 일반적으로 100MB 이상인 최소 데이터 블록을 스캔하는데, 이러한 시스템에서 초당 수천 개의 개별 항목을 조회하려고 하면 그 사용 사례에 맞게 설계된 캐싱 계층과 결합되지 않는 한 시스템이 중단될 것이다.

참고로, 초기 데이터 웨어하우스는 보통 트랜잭션 어플리케이션에 사용되는 것과 같은 RDBMS를 기반으로 구축되었었는데, MPP (Massively Parallel Processing) 시스템의 인기가 높아지면서 대용량 데이터에 걸쳐 검색 성능을 크게 개선할 수 있는 병렬 프로세싱으로 전환되었다고 한다.

구체적으로, OLTP 시스템에서 가장 많이 쓰이는 MySQL과 MongoDB, 그리고 OLAP 시스템 내에서 가장 인기있는 데이터 웨어하우스 솔루션인 BigQuery에 대한 심층적인 분석을 진행한 포스팅을 현재 게시했으니 관심이 있으시면 확인해주세요 :)

Data Discovery

사실, 데이터 엔지니어의 역할은 위에서 언급한 역할들에만 국한되지는 않습니다.

회사와 Data Team이 성장해가며 아래와 같은 시나리오가 그려질 수 있다.

Problem

1. 데이터가 커지면 테이블과 대시보드의 수가 증가 → 정보과잉의 문제
2. 데이터 분석 시 어느 테이블이나 대시보드를 봐야하는지 혼란 → 효율성의 문제
3. 데이터 변환 때문에 데이터 집합이 동일한 경로에서 어떻게 파생되었는지 알기 어려움

위와 같은 문제들은, Data Catalog 혹은 Data Discovery 문제로 치환될 수 있기 때문에, 데이터의 계보를 유지하고 모니터링할 수 있는 것도 성장된 데이터 팀에서의 데이터 엔지니어의 역할 중 하나입니다. 보통 아래와 같은 솔루션을 통해 해당 문제를 해결합니다.

Solution

1. 데이터 조회가 적은 사항에 대해 데이터 분석가와 협의하거나 자체적으로 사용 빈도를 추적하여, 주기적으로 테이블과 대시보드를 클린업 진행
2. Datahub나 Amundsen과 같이 Data Discovery에서 주로 발생하는 문제를 해결해주는 서비스를 이용

참고문헌

• 견고한 데이터 엔지니어링 (written by Joe Reis & Matt Housley)
• https://dovetail.com/product-development/data-driven-vs-data-informed/

AWS EC2 관련 스터디를 하던 중 IP에 대한 개념이 올바르게 정립되지 않은 것 같아 2학년 때 수강했던 컴퓨터 네트워크 전공책, 그리고 내가 알고 있는 개념을 정리하여 글을 적게 되었다. 일반적인(?) 상황에서 떠오르는 의문점을 해결하는 형식으로 글을 구성했다보니 IP에 대한 이해가 전무해도 흐름을 따라가다 보면 개념을 이해할 수 있을 것이다.

IP?

IP = Internet Protocol

인터넷에서 컴퓨터와 다른 장치가 통신할 때 사용되는 규약(프로토콜)이다. 또한 TCP/IP 프로토콜 스택에서 중요한 역할을 수행한다.

이해가 어렵다면 아래의 글을 통해 일단 IP를 인식하자.

네트워크 상에서 두 개의 컴퓨터끼리 통신을 한다고 가정한다면, 일단 그 둘을 식별하는 ID가 필요할 것이다. 그것이 바로 IP!

IP는 패킷 스위칭 네트워크에서 데이터를 전달하는 데 사용됩니다. 이때, IP는 송신자와 수신자 간의 주소를 지정하고 데이터를 작은 패킷으로 나누어서 전송한다. 이렇게 작게 분할된 패킷들은 최적의 경로로 전달되며, 도착지에서는 이를 재조합하여 완전한 데이터로 복원된다.

IPv4와 IPv6는 IP의 버전을 나타내며, 현재 대부분의 인터넷에서는 IPv4가 사용되고 있다. IPv4는 32비트 주소 체계를 사용하며, 이는 대략 42억개의 주소를 지원한다. 그러나 인터넷이 확장됨에 따라 IPv4 주소의 부족 문제가 발생하게 되어, 더 많은 주소를 지원하는 IPv6로의 전환도 이루어지고 있다.

직관적으로 본인이 이용하고 있는 네트워크에 대한 컴퓨터가 할당받은 IP를 보고 싶다면 명령 프롬프트에서 ipconfig를 실행하면 된다. (기본적인 설정에서도 확인가능)

Windows

ipconfig

mac OS

ip addr show

IP는 항상 하나로 고정?

IP는 고정IP와 동적IP의 두가지 형태로 존재한다.

본인이 장치에 대한 권한이 있거나 서버의 관리자라면 선택할 수 있다.

고정IP는 말 그대로 관리자가 수동으로 컴퓨터나 장치에 IP를 부여하는 방식이다.

쉽게 말하면 우리가 어떠한 서비스에 회원가입할 때 하나의 고정 ID를 생성하는 것과 같다. 그렇기에 장치가 off되더라도 이전에 설정했던 IP는 유지된다. 이러한 특성으로 보통 서버나 네트워크 장비에 이용한다.

동적IP는 DHCP서버가 IP주소를 컴퓨터나 장치에 자동으로 부여하는 방식이다.

쉽게 말하면 우리가 가입한 어떠한 서비스의 ID가 1주일에 한번씩 바뀐다는 것이다. (물론 자동으로 로그인이 될 것이다^^)

이처럼 네트워크 상에 연결될 때마다 DHCP서버에서 IP주소를 자동으로 할당하고 보통 스마트폰이나 개인용 컴퓨터에서 이용한다.

‘어떤 방식이 좋다!’ 와 같은 이해 방식 보다 어떤 네트워크에 연결되느냐, 본인의 장치가 어떤 목적과 특성을 가지고 있느냐에 초점을 맞추고 그에 따른 방식을 선택하는 것이 맞는 이해인듯 하다.

그렇다면 같은 네트워크를 써도 IP는 다르겠네?

그럴 확률이 높다.

물론 같은 네트워크에 연결된 장치들, 쉽게 말하면 가정용 Wifi 네트워크를 공유하고 있는 스마트폰, 테블릿이 고정IP 형식으로 동일한 IP주소를 할당받을 수 있다. 하지만 IP주소 충돌 (IP conflict)이슈가 발생한다.

통신 과정에서 IP주소가 같다면 충돌이 발생하고 서로의 패킷을 인식하지 못하는 이슈가 일어나기 때문에 DHCP서버에서 주소를 할당하는 동적IP방식을 채택하거나, 서로 다른 고정IP를 부여하는 방식으로 문제를 해결해야 한다.

정확히 IP를 부여하는 주체가 무엇이야?

무엇(DHCP 서버)일 수도 있고 누구(네트워크 관리자)일 수도 있다.

하지만 아마 이러한 질문이 나온다면 동적IP에 대한 물음일 것이라고 예상한다.

일반적으로 네트워크 장치에서 (Wifi라우터나 스위치)에서 부여한다. 주체를 정확히 하자면 앞서 언급한 DHCP(Dynamic Host Configuration Protocol) 서버이다.

DHCP 서버는 일련의 IP 주소 대역을 관리하며, 네트워크에 연결된 장치들이 DHCP 서버로부터 IP 주소를 요청하면, DHCP 서버는 이에 대해 유효한 IP 주소를 할당한다. 일반적으로 DHCP 서버는 할당된 IP 주소의 유효 기간(TTL, Time To Live)을 지정하여 일정 시간이 지나면 해당 IP 주소를 해제하고 다른 장치들에게 할당할 수 있도록 한다.

당연히 IP주소를 재할당하는 주기는 관리자가 설정할 수 있다.

또한, DHCP 서버는 일정한 규칙에 따라 IP 주소를 할당한다. 예를 들어, DHCP 서버에서는 할당 가능한 IP 주소 범위를 미리 설정하고, 클라이언트가 IP 주소를 요청할 때마다 사용 가능한 IP 주소 중 하나를 선택하여 할당한다. 이러한 알고리즘은 네트워크 관리자가 DHCP 서버를 구성할 때 설정할 수 있으며, 일반적으로 다양한 알고리즘이 지원된다.

동적IP 개념이 왜 나오게 된거야?

예상했겠지만 초기 인터넷 개발 환경에서는 동적IP개념이 없었지만 IP주소의 부족, 주소 관리의 효율성 등의 문제가 발생하여 나오게 된 개념이다.

이 밖에도 동적 IP방식을 사용하면 IP주소를 필요할 때만 할당하여 자원 효율성을 지킬 수 있고, 보안성, 비용 절감 등의 문제가 해결될 수 있다.

Problem: 서버에는 무조건 고정IP?

예전에는 서버를 구축한다고 하면 서버용 하드웨어를 직접 구매하여, 네트워크 카드를 설치하고 고정IP를 할당하곤 했다.

하지만 현재는 AWS 혹은 GCP에서 제공하는 클라우드 서비스를 이용하여 이러한 불필요한 문제를 해결할 수 있다.

어찌되었든 지금까지의 글을 이해했다면 당연히 서버는 고정IP 방식을 사용해야 한다. 서버가 꺼지던 관리자가 보수 작업을 하던, 클라이언트가 쉽게 다시 서버에 접속하기 위해서는 IP주소가 변하지 않아야 하기 때문이다.

AWS EC2 인스턴스에는 동적IP주소가 기본적으로 할당된다.

그렇다면 IP주소가 변경될 때마다 클라이언트들에게 변경된 IP주소를 알려줘야 하는 이슈가 발생하는데 이러한 문제를 어떻게 해결해야 할까?

Solution

1. AWS Elastic IP를 기능을 사용하면 EC2 인스턴스에 고정IP주소를 할당할 수 있다. 기본적으로 region당 5개까지 할당이 가능하고 추가로 필요하면 AWS 콘솔의 Support Center에서 Service limit increase를 진행해야 한다.

2. DNS (Domain Name System), AWS Route 53: DNS를 이용하여 인스턴스의 IP주소와 도메인 이름을 매핑하는 방식이다. 전통적으로 계속 사용하는 방식 중 하나인데, 클라이언트는 도메인 이름 (예: 웹사이트 주소)를 통해 접속하고, DNS서버는 이에 매핑되는 IP주소를 반환해 주는 방식이다. 따라서 고정되지 않는 IP문제를 해결할 수 있고 AWS같은 경우 AWS Route53 서비스가 DNS기능을 제공한다.

3. AWS ELB(Elastic Load Balancer): 다수의 인스턴스에 접속하는 클라이언트의 트래픽을 분산시킨다. 인스턴스의 IP주소가 변경되더라도 클라이언트는 ELB의 END-POINT에 접속하게 하며 문제를 해결한다.

서버의 관리자는 각각의 상황(비용, 인프라)에 맞게 방식을 채택해야 할 것이다.

DNS analysis in Wireshark

전공 수업 중 본인이 진행한 프로젝트 (Wireshark를 이용한 네트워크 분석) 자료이다. 당시 네이버 쇼핑 도메인을 이용하였다. DNS의 표준 포트 번호는 53번이다.

본인의 IPv4 주소 → 168.126.63.1(server) ## DNS request 진행

168.126.63.1(server)→ 본인의 IPv4 주소 ## DNS response 진행

고찰

_당연히 이러한 동적IP가 야기하는 문제점과 DNS와 같은 해결방식이 클라우드 서비스 때문에 발생한것이 아닌 것을 알고 있을 것이다. _

어떻게 보면 클라우드 서비스가 새로운 인터넷의 개념을 재정의 하거나 창조한 것이 아닐 수는 있다. 쉽게 말해서 Amazon이나 Google이 우리가 개별적으로 구축해야 할 서버용 하드웨어를 대신 대용량으로 구매하여 엄청난 크기의 데이터 센터를 구축한 뒤 우리에게 인프라를 제공하는 것이기 때문이다.

하지만 우리는 클라우드 서비스 덕분에 서버 구축에 필요한 초기 비용을 절감하고, 유연하게 자원을 할당할 수 있고, 계속해서 개발되는 매우 많은 서비스를 통해 부가 가치를 창출할 수 있다는 점에서 클라우드 서버 구축 환경은 엄청난 혁신이라고 생각한다.

참고문헌

1. 컴퓨터 네트워킹: 하향식 접근 (By James Kurose, Keith Ross)
2. https://aws.amazon.com/ko/what-is/computer-networking/
3. https://ko.wikipedia.org/wiki/인터넷_프로토콜

본 글은 Django framework 공식문서에서 소개하는 설치 과정과 제시하는 MTV패턴을 중심으로 tutorial의 핵심 요소를 정리한 글입니다.

What is Django?

Django는 파이썬으로 작성된 고수준 웹 프레임워크로, 신속한 개발과 깔끔한 디자인을 가능하게 합니다. Django는 "batteries-included" 철학을 따르며, 많은 기능을 내장하여 웹 개발에 필요한 거의 모든 것을 제공합니다.

Django는 MTV 패턴을 따릅니다. MTV는 Model-Template-View의 약자로, 흔히 우리가 아는 MVC(Model-View-Controller) 패턴의 변형입니다. Django에서는 View가 실제로 비즈니스 로직을 처리하는 것이 아니라, 템플릿 시스템을 사용하여 데이터를 표시하는 역할을 합니다.

• Model: 데이터베이스 구조를 정의합니다. 데이터의 필드와 동작을 정의하며, 데이터베이스와의 상호 작용을 관리합니다.
• Template: 사용자에게 표시될 화면을 정의합니다. HTML과 Django 템플릿 언어를 사용하여 데이터를 렌더링합니다.
• View: 요청을 처리하고 적절한 템플릿을 선택하여 응답을 반환합니다. 비즈니스 로직을 구현하고, 모델과 템플릿을 연결합니다.

이제 공식문서에서 제시하는 가이드라인에 따라 Django를 설치해보고, 간단하게 구성 요소를 살펴봅시다.

초기 세팅

가상환경

기존 환경에서 실행해도 상관은 없지만 여러 프로젝트의 버젼 충돌로 인한 에러를 방지하기 위하여 virtual environment를 구축하는것이 바람직하다

가상환경 생성

py -m venv project-name

가상환경 활성화

project-name\Scripts\activate.bat

가상환경 비활성화

deactivate

django 설치

py -m pip install Django

장고 버젼 확인

django-admin –version

장고 프로젝트 생성

$ django-admin startproject mysite

장고 서버 실행

$ python manage.py runserver

앱 생성

$ python manage.py startapp polls ## polls -> App 이름

생성된 앱을 서버에서 구축하기 위해서는 urls path와 settings를 관리해야 합니다.

참고로, 아래에서 언급되는 urls.py파일은 쉽게 설명하면 API request path를 관리하는 파일이라고 보면 됩니다. (ex: Node.js 프로젝트에서 관리하는 index.ts 파일)

프로젝트 디렉토리의 urls.py

from django.contrib import admin
from django.urls import path, include
urlpatterns = [
    path("admin/", admin.site.urls),
    path("polls/", include('polls.urls'))
]

특정 앱 디렉토리 (ex: polls)의 urls.py

from django.urls import path
from . import views
urlpatterns = [
    path('',views.index, name='index')
    path('some_url',views.some_url)
]

프로젝트 디렉토리의 settings.py

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'polls.apps.PollsConfig',
]

Models

장고의 model은 RDB의 개념이라고 생각하면 쉽다. 따라서 SQL과 같이 schema를 생성하고 attribute을 추가하고 이에 대한 filtering이 가능하다.

앱 디렉토리의 models.py

from django.db import models

class Question(models.Model):
    question_text = models.CharField(max_length=200)
    pub_date = models.DateTimeField('date published')

class Choice(models.Model):
    question = models.ForeignKey(Question, on_delete=models.CASCADE)
    choice_text = models.CharField(max_length=200)
    votes = models.IntegerField(default=0)

Templates

화면단 구현에 필요한 각각의 html 파일은 template 디렉토리에서 보통 관리한다.

index.html 예시

{% if questions %}
<ul>
    {% for question in questions %}
        <li>{{question}}</li>
    {% endfor %}
</ul>
{% else %}
<p>no questions</p>
{% endif %}

Views

views.py

각각의 html파일을 함수 단위로 manage한다.

from .models import *
from django.shortcuts import render

def index(request):
    latest_question_list = Question.objects.order_by('-pub_date')[:5]
    context = {'questions': latest_question_list}
    return render(request, 'polls/index.html', context)

def detail(request, question_id):
    question = Question.objects.get(pk=question_id)
    return render(request, 'polls/detail.html', {'question': question})

참고로, render는 간단하게 말하면 html파일을 화면에 보이게 만들어 주는 것이라고 이해하면 된다.

urls.py

추가한 view의 결과를 보기 위해서는 path를 추가해야한다

from django.urls import path 
from . import views  

app_name = 'polls'

urlpatterns = [     
    path('', views.index, name='index'),
    path('some_url', views.some_url), 
    path('<int:question_id>/', views.detail, name='detail'),     
]

Error 처리

404에러: http에서 요청한 페이지를 찾을 수 없을 때 생기는 에러

404 에러 방지

def detail(request, question_id):

    try:
        question = Question.objects.get(pk=question_id)
    except Question.DoesNotExist:
        raise Http404("Question does not exist")

    question = get_object_or_404(Question, pk=question_id)
    return render(request, 'polls/detail.html', {'question': question})

서버에서 요청을 동시에 처리할 때 발생할 수 있는 에러 방지

Solve: 두개의 서버가 아닌 하나의 DB에서 처리

def vote(request, question_id):
    question = get_object_or_404(Question, pk=question_id)
    try:
        selected_choice = question.choice_set.get(pk=request.POST['choice'])
    except (KeyError, Choice.DoesNotExist):
        return render(request, 'polls/detail.html', {'question': question, 'error_message': f"선택이 없습니다. id={request.POST['choice']}"})
    else:
        selected_choice.votes = F('votes') + 1
        selected_choice.save()
        return HttpResponseRedirect(reverse('polls:index'))

Admin

• django에는 관리자 페이지 및 관리자 기능을 할 수 있는 admin.py를 제공한다

• 관리자 페이지에서 보여질 모델의 필드, 필터링 기준, 검색 기준 등을 설정할 수 있으며, 모델에 대한 CRUD(CREATE, READ, UPDATE, DELETE) 작업을 수행하는 메소드도 정의할 수 있다.
• 쉽게 말하면 Admin을 활용하여 DB의 내용 관리(CRUD)를 손쉽게 할 수 있다는 것이다.

CRUD를 위한 Admin page 커스터마이즈

from django.contrib import admin
from .models import Choice, Question

admin.site.register(Choice)

class ChoiceInline(admin.TabularInline):
    model = Choice
    extra = 3


class QuestionAdmin(admin.ModelAdmin):
    fieldsets = [
        ('질문 섹션', {'fields': ['question_text']}),
        ('생성일', {'fields': ['pub_date'], 'classes': ['collapse']}),        
    ]
    readonly_fields = ['pub_date']
    inlines = [ChoiceInline]
    list_filter = ['pub_date']
    search_fields = ['question_text', 'choice__choice_text']

admin.site.register(Question, QuestionAdmin)

지금까지 공식문서를 따라, Django 아키텍처 내에서 MTV패턴이 대략적으로 어떤 구조로 구현되고, 에러에 대한 관리와 admin 기능을 어떻게 코드로 관리하는지에 대해 알아보았다.

이후, tutorial에서 제시하는 가이드라인을 전체 구현한 코드는 아래 github repository에 업로드 해놓았습니다 :) GitHub - idle-danie/django_study: django_study_tutorial

참고문헌

https://www.djangoproject.com/start/

디지털 마케팅은 데이터 기반 마케팅을 뜻한다

접점 = 채널 = 광고미디어

접점 데이터 수집 및 저장하는 것은 매우 중요한 요소이다.

• 고객은 다양한 경로를 통해 방문, 구매를 진행
• 가격이 상대적으로 비싼 물건이나 서비스일수록 시간을 두고 여러경로를 통해 같은 사이트를 여러 번 방문하면서 리서치
• 방문에 기여한 채널들을 기록하는 것이 중요

최종 전환 기록

이는 구매나 회원가입과 같은 마케팅의 목표를 위해 필요한 기록이다.

보조 전환 기록

• 보조전환은 최종전환의 징조
• 사용자의 방문 정보들이 계속 추출되어야 함

_위와 같은 수집활동 데이터가 저장되는 데이터베이스가 바로 디지털 마케팅 데이터 인프라의 핵심! _

분석 방법

1. 라스트 터치
2. 퍼스트 터치
3. 멀티 터치

Last touch model

• 보통 싼 물건
• 최종 구매전의 마지막 채널에게 모든 성과

Last non-direct touch model

• 마지막 채널이지만, 직접방문이 아니느 그 전 채널에게 성과를 부여

First

• 처음 방문 채널

Linear (멀티 터치 모델)

• 모든 채널에게 동일하게 나눠주는 모델

Time decay (멀티 터치 모델)

• 가장 최근모델들에게 가중치를 더 주는 모델

고객의 평생 가치 (Life time balue)

• 사용자의 초기행동을 보고 미래에 가져다줄 수 있는 가치 예측
• 처음에는 간단하게 휴리스틱 기반 예측 가능

고객 이탈률 (customer churn) 예측 (리텐션과 반대)

• 리타게팅 광고: 쿠키를 사용하여 유저를 따라다니며 광고 제시
• 쿠키: 점점 사용할 수 없는 방향으로 법안 적용 -> 정교한 머신러닝 모델로 대체

마케팅 핵심 지표

개인적으로 마케팅을 진행할 때, 데이터 엔지니어라도 필수적으로 이해하고 있어야 한다고 생각하는 지표 14가지의 정의와 계산 방법을 정리해보았다.

1. Cost:
   • 설명: 광고 캠페인에 소요된 총 비용.
     • 계산 방법: 플랫폼에서 제공하는 광고 비용 데이터의 합계.
2. Impressions:
   • 설명: 광고가 사용자에게 노출된 총 횟수.
   • 계산 방법: 플랫폼에서 제공하는 노출 수 데이터의 합계.
3. Click:
   • 설명: 광고가 클릭된 총 횟수.
   • 계산 방법: 플랫폼에서 제공하는 클릭 수 데이터의 합계.
4. Install:
   • 설명: 광고를 통해 앱이 설치된 총 횟수.
   • 계산 방법: 플랫폼에서 제공하는 설치 수 데이터의 합계.
5. Sign-up:
   • 설명: 광고를 통해 사용자가 가입한 총 횟수.
   • 계산 방법: 플랫폼에서 제공하는 가입 수 데이터의 합계.
6. Purchase:
   • 설명: 광고를 통해 구매가 이루어진 총 횟수.
   • 계산 방법: 플랫폼에서 제공하는 구매 수 데이터의 합계.
7. Revenue:
   • 설명: 광고를 통해 발생한 총 수익.
   • 계산 방법: 플랫폼에서 제공하는 수익 데이터의 합계.
8. CTR (Click-Through Rate):
   • 설명: 광고 노출 수 대비 클릭 수 비율.
   • 계산 방법: (클릭 수 / 노출 수) * 100
9. CPC (Cost Per Click):
   • 설명: 클릭당 비용.
   • 계산 방법: 광고 비용 / 클릭 수
10. CPI (Cost Per Install):
    • 설명: 설치당 비용.
    • 계산 방법: 광고 비용 / 설치 수
11. CPA (Cost Per Action) - Sign-up:
    • 설명: 가입당 비용.
    • 계산 방법: 광고 비용 / 가입 수
12. CAC (Customer Acquisition Cost) - Sign-up:
    • 설명: 고객 한 명을 획득하는 데 드는 비용.
    • 계산 방법: 광고 비용 / 고객 획득 수 (여기서는 가입 수)
13. CPA (Cost Per Action) - Purchase:
    • 설명: 구매당 비용.
    • 계산 방법: 광고 비용 / 구매 수
14. ROAS (Return On Ad Spend):
    • 설명: 광고 비용 대비 수익 비율.
    • 계산 방법: (광고 수익 / 광고 비용) * 100

SQLD를 공부하면서, SQL의 basic concept와 주의하거나 알면 좋은 지점들에 대해서 정리해보았습니다.

What is SQL?

SQL(Structured Query Language)은 관계형 데이터베이스 관리 시스템(RDBMS)에서 데이터를 관리하고 조작하기 위해 사용되는 표준 언어입니다. SQL을 사용하면 데이터베이스에 저장된 데이터를 삽입, 수정, 삭제 및 조회할 수 있습니다. SQL은 강력한 데이터 조작 기능과 함께 데이터베이스 스키마 생성 및 변경을 위한 명령도 제공합니다. SQL의 주요 구성 요소로는 DDL(데이터 정의 언어), DML(데이터 조작 언어), DQL(데이터 질의 언어) 등이 있습니다.

SQL 입문자를 위한 Tip

• 다수의 SQL문은 세미콜론으로 분리합니다.
• SQL 주석
  • --: 인라인 한 줄짜리 주석
  • /* ... */: 여러 줄에 걸쳐 사용 가능한 주석
• SQL 키워드는 나름대로의 포맷팅이 필요합니다(팀끼리 상의하여 컨벤션을 정하는 것이 중요)
  • 테이블 및 필드 이름 명명 규칙 정하기
    • 단수형 vs 복수형 예) User vs Users
    • _ vs CamelCasing 예) user_time vs UserTime

SQL vs Pandas

Pandas가 등장할 당시에는, '전반적인 데이터 처리 트렌드가 SQL에서 Pandas로 넘어오겠다'라고 생각하는 개발자들도 존재했다고 한다. 하지만, SQL과 달리 코드 관계자가 아닌 인원이 코드를 보고 한눈에 어떤 의미인지를 모른다는 치명적인 단점이 존재하여, 여전히 SQL은 데이터 조작 언어로서 입지를 공공히 하고 있다.

물론 설계 목적 자체가 다르기 때문에 비교하는 것이 무의미할 수 있지만, 데이터를 다룬다는 큰 범주안에서 유사한 역할을 담당한다고 볼 수 있기에 같이 설명을 드리려고 합니다.

• SQL: 관계형 데이터베이스에 저장된 데이터를 효율적으로 관리하고 질의할 수 있도록 설계된 언어입니다.
• Pandas: Python 환경에서 데이터 분석 및 조작을 위해 설계된 라이브러리로, 데이터프레임을 사용하여 다양한 데이터 조작을 간편하게 수행할 수 있습니다.

보통 SQL은 디스크 기반의 대규모 데이터베이스에서 효율적으로 작동하고, 데이터베이스 서버에서 실행됩니다. 반면에 Pandas는 데이터프레임 형태로 메모리에 데이터를 로드하여 빠르고 유연한 데이터 조작을 가능하게 합니다. 사실, Pandas의 메모리 할당 방식은 대규모 데이터를 다루기 어렵기 때문에, 이 때 유사한 용도로 PySpark가 많이 쓰입니다. 대규모 데이터에 대한 명확한 기준은 없지만, 경험 상 10GB가 넘어가는 데이터를 처리할 때는 PySpark를 사용하는 것이 유리한 것 같습니다.

DDL

CREATE TABLE

• primary key 속성을 지정할 수 있으나 무시됩니다.
• CTAS:
  • CREATE TABLE ... AS SELECT ...을 통해 서머리 테이블을 생성할 수 있습니다.
  • ELT의 역할을 할 수 있지만, 테스트할 수 없고 테이블 컬럼에 대한 디테일한 관리가 어렵습니다.
  • 초기 설정한 데이터 타입이 최종 타입입니다.
  • CREATE TABLE → INSERT → SELECT
  • 보통 DBT(Data Build Tool)를 사용합니다.

DROP TABLE

• DROP TABLE ...;:
  • 없는 테이블을 지우려고 하면 에러가 발생합니다.
• DROP TABLE IF EXISTS ...; 사용을 권장합니다.
• VS DELETE FROM:
  • DELETE FROM은 조건에 맞는 레코드들을 삭제합니다(테이블 자체는 유지).

ALTER TABLE

• 새로운 컬럼 추가:
  • ALTER TABLE 테이블이름 ADD COLUMN 필드이름 필드타입;
• 기존 컬럼 이름 변경:
  • ALTER TABLE 테이블이름 RENAME 현재필드이름 TO 새필드이름;
• 기존 컬럼 제거:
  • ALTER TABLE 테이블이름 DROP COLUMN 필드이름;
• 테이블 이름 변경:
  • ALTER TABLE 현재테이블이름 RENAME TO 새테이블이름;

DML

SELECT, FROM, WHERE 절은 너무 기초적인 문법이기에 설명에서 제외

IN

• WHERE channel IN ('Google', 'Youtube'):
  • WHERE channel = 'Google' OR channel = 'Youtube'

LIKE(구별) and ILIKE (대소문자 구별 안 함)

• WHERE channel LIKE 'G%' → 'G*'
• WHERE channel LIKE '%o%' → '*o*'
• NOT LIKE 또는 NOT ILIKE
  • mysql은 대소문자 구별을 하지 않음
  • PostgreSQL/Redshift는 구별함

BETWEEN

• DATE RANGE MATCHING

String functions

• LEFT(str, N)
• REPLACE(str, exp1, exp2)
• UPPER(str)
• LOWER(str)
• LEN(str)
• LPAD, RPAD
• SUBSTRING

INSERT INTO VS .COPY

• 일반적으로 INSERT가 더 느립니다. (배치 삽입 메커니즘 이해 필요)
• INSERT INTO table SELECT * FROM ...:
  • 필드의 타입을 제어하려면 CREATE TABLE table AS SELECT보다 낫습니다.
  • 그러나 varchar 길이를 맞추는 것은 쉽지 않습니다.
  • Snowflake와 BigQuery는 string 타입을 지원합니다.

GROUP BY

• DAU, WAU, MAU 계산할 때 GROUP BY가 필요합니다.

  --- mau 계산 sql예시
  SELECT TO_CHAR(A.TS, 'YYYY-MM') AS month,
         COUNT(DISTINCT B.userid) AS mau
  FROM raw_data.session_timestamp A
  JOIN raw_data.user_session_channel B ON A.sessionid = B.sessionid
  GROUP BY 1
  ORDER BY 1 DESC;
  

ORDER BY

• NULL value ordering (NULL이 가장 큰 값?)
  • In Redshift, NULL은 최대값으로 간주됩니다.
    • ORDER BY 1 DESC; → NULL이 가장 앞에 위치
    • ORDER BY 1 DESC NULLS LAST; → NULL이 맨 뒤로 이동
• ORDER BY와 GROUP BY → 포지션 번호 vs 필드 이름
  • GROUP BY 1 == GROUP BY month == GROUP BY TO_CHAR(A.ts, 'YYYY-MM')

Type Cast and Conversion

• Type casting
  • cast 또는 :: 연산자 사용
    • channel::int (본 예시는 PostgreSQL기반의 Redshift에 해당, SQL마다 다를 수 있음)
    • cast(channel as int)
• Conversion
  • Date conversion
    • convert_timezone
      • convert_timezone('America/Los_Angeles', ts)
      • SELECT pg_timezone_names()
    • date, truncate
    • date_trunc
      • 첫 번째 인자가 어떤 값을 추출하는지 지정
    • extract 또는 date_part: 날짜, 시간에서 특정 부분의 값을 추출
    • datediff, dateadd, get_current...
  • TO_CHAR(A.TS, 'YYYY-MM') AS month은 아래와 같이 사용해도 같은 output을 도출
    • LEFT(A.ts, 7)
    • DATE_TRUNC('month', A.ts)
    • SUBSTRING(A.TS, 1, 7)

NULL

값이 존재하지 않음을 의미하고, 0과 비어있는 문자열과는 다르다는 점을 인지해야 한다.

• IS NULL, IS NOT NULL 형식으로 사용
  • Boolean 타입의 필드도 IS TRUE, IS FALSE 형식으로 비교
• LEFT JOIN 시 매칭되는 것이 있는지 확인할 때 유용
• NULL 값을 다른 값으로 변환하고 싶다면?
  • COALESCE
  • NULLIF
• 특정 값을 NULL이나 0으로 나누면?
  • NULL로 나누면 결과는 NULL

    SELECT 10 / NULL; -- 결과: NULL

  • 0으로 나누면 오류(Division by zero error)가 발생

    SELECT 10 / 0; -- 결과: 오류 (Division by zero)
    ## Count
    | … | … | value |
    | --- | --- | --- |
    | … | … | NULL |
    | … | … | 1 |
    | … | … | 1 |
    | … | … | 0 |
    | … | … | 0 |
    | … | … | 4 |
    | … | … | 3 |
    

COUNT(0) FROM Table = 7 
COUNT(value) FROM Table = 6 
COUNT(DISTINCT value) FROM Table = 4

SQL 실행순서

세부적으로 들어가면 사실 최종적인 실행순서는 보통 쿼리 옵티마이져에 의해서 결정된다. 다만, 기본적으로 아래 사진과 같이 이해하고 있다면 SQL문을 작성할 때 크게 도움이 된다.

참고문헌

https://jaehoney.tistory.com/191

Colab을 통해 Redshift로의 connection 과정에서 문제가 발생했다. 1주일 전까지는 이러한 문제가 발생되지 않았기에 내가 관리하지 않은 웹상 개발환경인 colab에서의 최근에 일어난 update과정이 문제가 되었다고 추론했다.

Solve

python ORM인 SqlAlchemy가 새로운 버전으로 업데이트 되는 과정에서 충돌이 발생한 것으로 보인다.

https://www.sqlalchemy.org/download.html 따라서 이전 버전인 1.4.47을 설치하였고 정상적으로 작동하는 것을 확인했다.

파이썬 노트북 환경에서는 이러한 일이 비일비재하고 큰 변화없이 모듈이 실행이 안된다면 모듈의 업데이트 현황을 확인하는 것이 습관이 되었다.

오래 전 자료구조 시간에 배운 Linked List에 대한 개념이 머릿속에서 모호해져서, 파이썬으로 직접 구현하며 개념을 상기해보려 한다.

Linked list

What is Linked list?

Linked list(연결 리스트)는 데이터를 노드(Node)라는 단위로 저장하는 자료구조다. 각 노드는 데이터 필드와 다음 노드를 가리키는 포인터로 구성되어 있으며, 이러한 노드들이 일렬로 연결되어 있는 형태를 띄고 있다. Linked list의 주요 특징은 삽입과 삭제가 용이하다는 점이다. 배열과 달리 미리 할당된 메모리 공간이 필요 없으며, 필요한 만큼 메모리를 할당받을 수 있다.

Linked list는 주로 다음과 같은 연산을 지원한다:

• 삽입(Insertion): 리스트의 앞, 중간, 끝에 새로운 노드를 삽입할 수 있음
• 삭제(Deletion): 특정 노드를 리스트에서 제거할 수 있음
• 탐색(Search): 리스트를 순회하며 특정 값을 가진 노드를 찾을 수 있음

Code implementation

linked list를 파이썬으로 직접 구현하면 아래와 같을 것입니다 ㅎ

class Node:
    def __init__(self, val):
        self.val = val
        self.next = None


class SingleLinkedList:
    def __init__(self):
        self.head = None

    def insertAtHead(self, val): 
        node = ListNode(val)
        node.next = self.head
        self.head = node

    def insertBack(self, val): 
        node = ListNode(val)
        crnt_node = self.head
        while crnt_node.next:
            crnt_node = crnt_node.next
        crnt_node.next = node

    def findNode(self, val): 
        crnt_node = self.head
        while crnt_node is not None:
            if crnt_node.val == val:
                return crnt_node
            crnt_node = crnt_node.next
        raise RuntimeError('LinkedList: empty')

    def insertAfter(self, node, val): 
        new_node = ListNode(val)
        new_node.next = node.next
        node.next = new_node

    def popAfter(self, prev_node): 
        if prev_node.next is not None:
            prev_node.next = prev_node.next.next

간단하게 코드에 대한 설명을 하자면 아래와 같습니다.

• Node 클래스는 연결 리스트의 각 노드를 나타냅니다. 각 노드는 값(val)과 다음 노드를 가리키는 포인터(next)를 가집니다.
• SingleLinkedList 클래스는 단일 연결 리스트를 나타냅니다.
• insertAtHead 메서드는 리스트의 맨 앞에 새로운 노드를 삽입합니다.
• insertBack 메서드는 리스트의 맨 뒤에 새로운 노드를 삽입합니다.
• findNode 메서드는 리스트를 순회하며 특정 값을 가진 노드를 찾습니다.
• insertAfter 메서드는 특정 노드 뒤에 새로운 노드를 삽입합니다.
• popAfter 메서드는 특정 노드 뒤의 노드를 삭제합니다.

Double Linked list

What is Double Linked list?

Double Linked list(이중 연결 리스트)는 각 노드가 두 개의 포인터를 가지는 자료구조다. 하나는 다음 노드를 가리키고 다른 하나는 이전 노드를 가리킨다. 이러한 구조 덕분에 양방향으로 리스트를 순회할 수 있어 단일 연결 리스트에 비해 노드의 삽입과 삭제가 더 효율적이다.

Double Linked list의 주요 특징은 다음과 같다.

• 양방향 순회: 리스트의 앞과 뒤 어느 방향으로든 순회가 가능
• 더 빠른 삽입과 삭제: 특정 노드의 앞이나 뒤에 노드를 삽입하거나 삭제할 때, 추가적인 포인터 조작이 필요 없음

Code implementation

double linked list를 파이썬으로 직접 구현하면 아래와 같을 것입니다 ㅎ

class Node:

    def __init__(self, item):
        self.data = item
        self.prev = None
        self.next = None


class DoublyLinkedList:

    def __init__(self):
        self.nodeCount = 0
        self.head = Node(None)
        self.tail = Node(None)
        self.head.prev = None
        self.head.next = self.tail
        self.tail.prev = self.head
        self.tail.next = None


    def __repr__(self):
        if self.nodeCount == 0:
            return 'LinkedList: empty'

        s = ''
        curr = self.head
        while curr.next.next:
            curr = curr.next
            s += repr(curr.data)
            if curr.next.next is not None:
                s += ' -> '
        return s


    def getLength(self):
        return self.nodeCount


    def traverse(self):
        result = []
        curr = self.head
        while curr.next.next:
            curr = curr.next
            result.append(curr.data)
        return result


    def reverse(self):
        result = []
        curr = self.tail
        while curr.prev.prev:
            curr = curr.prev
            result.append(curr.data)
        return result


    def getAt(self, pos):
        if pos < 0 or pos > self.nodeCount:
            return None

        if pos > self.nodeCount // 2:
            i = 0
            curr = self.tail
            while i < self.nodeCount - pos + 1:
                curr = curr.prev
                i += 1
        else:
            i = 0
            curr = self.head
            while i < pos:
                curr = curr.next
                i += 1

        return curr


    def insertAfter(self, prev, newNode):
        next = prev.next
        newNode.prev = prev
        newNode.next = next
        prev.next = newNode
        next.prev = newNode
        self.nodeCount += 1
        return True


    def insertAt(self, pos, newNode):
        if pos < 1 or pos > self.nodeCount + 1:
            return False

        prev = self.getAt(pos - 1)
        return self.insertAfter(prev, newNode)


    def popAfter(self, prev):
        curr = prev.next
        next = curr.next
        prev.next = next
        next.prev = prev
        self.nodeCount -= 1
        return curr.data


    def popAt(self, pos):
        if pos < 1 or pos > self.nodeCount:
            raise IndexError('Index out of range')

        prev = self.getAt(pos - 1)
        return self.popAfter(prev)


    def concat(self, L):
        self.tail.prev.next = L.head.next
        L.head.next.prev = self.tail.prev
        self.tail = L.tail

        self.nodeCount += L.nodeCount

이번에도 간단하게 코드에 대한 설명을 하자면 아래와 같습니다.

• Node 클래스는 이중 연결 리스트의 각 노드를 나타냅니다. 각 노드는 값(data), 이전 노드를 가리키는 포인터(prev), 다음 노드를 가리키는 포인터(next)를 가집니다.
• DoublyLinkedList 클래스는 이중 연결 리스트를 나타냅니다.
• __repr__ 메서드는 리스트의 모든 노드를 문자열로 반환합니다.
• getLength 메서드는 리스트의 길이를 반환합니다.
• traverse 메서드는 리스트를 순회하며 모든 노드의 값을 리스트로 반환합니다.
• reverse 메서드는 리스트를 역순으로 순회하며 모든 노드의 값을 리스트로 반환합니다.
• getAt 메서드는 특정 위치에 있는 노드를 반환합니다.
• insertAfter 메서드는 특정 노드 뒤에 새로운 노드를 삽입합니다.
• insertAt 메서드는 특정 위치에 새로운 노드를 삽입합니다.
• popAfter 메서드는 특정 노드 뒤의 노드를 삭제합니다.
• popAt 메서드는 특정 위치의 노드를 삭제합니다.
• concat 메서드는 두 리스트를 연결합니다.

마무리

실제로 해당 개념들이 어떤 상황에서 활용되면 좋은지 알아보자 :)

Linked List 활용

1. 동적 메모리 할당: 배열과 달리 미리 크기를 정해놓지 않아도 되기 때문에, 동적으로 크기가 변하는 데이터를 처리할 때 유용하다. 예를 들어, 메모리 관리, 객체 풀(pool) 등에서 사용된다.

2. 데이터 삽입/삭제가 빈번한 경우: 특정 위치에 데이터를 삽입하거나 삭제할 때 배열보다 효율적이다. 삽입/삭제 시에 모든 요소를 이동시킬 필요 없이 포인터만 조작하면 되기 때문이다.

3. 스택과 큐 구현: Linked List는 스택과 큐 같은 자료구조를 구현하는 데 자주 사용된다. 스택에서는 후입선출(LIFO), 큐에서는 선입선출(FIFO) 방식으로 데이터를 처리할 수 있다.

4. 그래프 및 트리 구현: 그래프와 트리 같은 복잡한 자료구조도 Linked List를 사용해서 구현할 수 있다. 특히, 각 노드가 여러 자식 노드를 가질 수 있는 상황에서 유용하다.

Double Linked List 활용

1. 양방향 순회: 이중 연결 리스트는 앞뒤로 자유롭게 순회할 수 있어서, 양방향 탐색이 필요한 경우에 적합하다. 예를 들어, 뒤로 가기/앞으로 가기 기능이 있는 웹 브라우저의 히스토리 관리에 사용된다.

2. 이중 연결 리스트 기반 자료구조: Deque(양쪽 끝에서 삽입과 삭제가 가능한 큐), LRU(Least Recently Used) 캐시 등에서 사용된다. LRU 캐시는 최근에 사용된 적이 없는 데이터를 제거하는 방식의 캐시 알고리즘인데, 이중 연결 리스트를 사용하면 효율적으로 구현할 수 있다.

3. 텍스트 편집기: 텍스트 편집기에서 커서의 이동, 삽입, 삭제 같은 연산을 빠르게 처리하기 위해 이중 연결 리스트를 사용한다. 커서가 문장의 중간에 있을 때도 효율적으로 삽입/삭제가 가능하기 때문이다.

이처럼 Linked List와 Double Linked List는 각각의 장점을 활용하면, 다양한 상황에서 효율적인 데이터 처리를 가능하게 할 수 있다.

KEYWORD: 하드웨어 개발자와 소프트웨어 개발자와의 관계

• 1st party -> 하드웨어 개발자, 원천기술 보유자 ex) Apple, MS
• 2nd party -> 하드웨어 개발자와의 직접적인 관계를 통해 소프트웨어를 개발하는 자 ex) 1st party와 하청관계를 가지고 개발하는 자
• 3rd party -> 하드웨어 개발자와의 직접적인 관계 없이 소프트웨어를 개발하는 자, 원천기술을 활용하여 개발하는 자, 1st party와 user를 연결해주는 자 ex) Apple Appstore에서 앱 출시하는 앱 개발자, 플러그인 & 라이브러리 & 프레임워크 개발자

비즈니스적인 측면에서도 3rd party는 1st party의 프로덕트를 이용하게 하기 때문에 공생 관계라고 볼 수 있다

구체적으로 공생 관계에 대한 예시는 아래와 같을 것입니다.

1. 앱스토어 생태계: Apple의 Appstore는 3rd party 개발자들이 만든 수많은 앱들로 가득 차 있습니다. 이들은 Apple의 iOS 플랫폼을 활용하여 혁신적인 앱을 개발하고, 사용자들에게 다양한 경험을 제공합니다. 이는 iOS 플랫폼의 가치를 높이고, 사용자들이 Apple 제품을 계속 사용하게 만드는 요인 중 하나입니다.
2. 플러그인 및 라이브러리: 소프트웨어 개발에서 3rd party가 개발한 플러그인과 라이브러리는 개발자들이 보다 쉽게 기능을 추가하고, 효율적으로 작업할 수 있도록 돕습니다. 예를 들어, Python의 Pandas 라이브러리는 데이터 엔지니어들이 데이터 처리 작업을 더욱 간편하게 수행할 수 있도록 해줍니다.

참고문헌 https://ko.m.wikipedia.org/wiki/%EC%84%9C%EB%93%9C_%ED%8C%8C%ED%8B%B0_%EA%B0%9C%EB%B0%9C%EC%9E%90

enumerate(list): 원소와 인덱스 동시에 얻을 수 있는 함수

for i, element in enumerate(a, start = 1):
    print(i, element)

1 daniel
2 john
3 alex

zip(): iterator 객체 사용하여 병렬처리 가능

>>> name = ["Daniel", "Alex", "Steve"]
>>> st_num = [101, 102, 103]
>>> st_score = [100, 90, 34]
>>> st_score.append(12)
exam_score = list(zip(name, st_num, st_score))
>>> exam_score
[('Daniel', 101, 100), ('Alex', 102, 90), ('Steve', 103, 34)]

>>> a, b, c = zip(*exam_score)
>>> a, b, c
(('Daniel', 'Alex', 'Steve'), (101, 102, 103), (100, 90, 34))
>>> exam_score
[('Daniel', 101, 100), ('Alex', 102, 90), ('Steve', 103, 34)]

>>> dict(zip(name, st_num))
{'Daniel': 101, 'Alex': 102, 'Steve': 103}

>>> for i in zip(name, st_num):
    print(i[0], i[1])

Daniel 101
Alex 102
Steve 103

from collections import Counter: 딕셔너리 형의 객체를 반환하여 요소의 개수를 value로 확인할 수 있다. most_common()은 tuple이 포함된 리스트를 반환한다.

>>> from collections import Counter
>>> counter = Counter() # 빈 카운터 생성

>>> Counter(name)
Counter({'Daniel': 1, 'Alex': 1, 'Steve': 1})
>>> Counter("sdifjodvmef")
Counter({'d': 2, 'f': 2, 's': 1, 'i': 1, 'j': 1, 'o': 1, 'v': 1, 'm': 1, 'e': 1})

>>> a = Counter(name)
>>> a["Daniel"]
1

>>> Counter("doijfoivmkd").most_common()
[('d', 2), ('o', 2), ('i', 2), ('j', 1), ('f', 1), ('v', 1), ('m', 1), ('k', 1)]
>>> Counter("udsinjckwdckomvl").most_common(3)
[('d', 2), ('c', 2), ('k', 2)]

>>> b = Counter([1,1,1,1,1,1,1,1,2,2,2,12,31])
>>> a+b
Counter({1: 13, 2: 8, 21: 1, 12: 1, 31: 1})
>>> a-b
Counter({2: 2, 21: 1})
>>> a=b.total()
Traceback (most recent call last):
  File "<pyshell#41>", line 1, in <module>
    a=b.total()
AttributeError: 'Counter' object has no attribute 'total'
파이썬 3.1이후에 total()이 추가되었다고 하는데 지금은 호출되지 않는다. 굳이 구하려 한다면 아래와 같이 구하면 될 것이다. 
>>> sum(a.values())
11

>>> a & b
Counter({1: 5, 2: 3})
>>> a | b
Counter({1: 8, 2: 5, 21: 1, 12: 1, 31: 1})
>>> a.items()
dict_items([(1, 5), (2, 5), (21, 1)])
>>> a.values()
dict_values([5, 5, 1])

>>> list(a.elements())
[1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 21]

>>> a.update([1,1,1,1,1,1,1])
>>> a
Counter({1: 12, 2: 5, 21: 1})

>>> a.subtract([2,2,2,2])
>>> 
>>> a
Counter({1: 12, 2: 1, 21: 1})

*sorted의 key활용: 다중조건에서 오름차순을 사용할 때 str같은 경우 -를 이용할 때 bad operand type으로 취급된다. *

>>> sorted(array, key = lambda x:(x[1], x[0]))
[('b', 2), ('g', 3), ('a', 4), ('k', 4), ('f', 5)]
>>> sorted(array, key = lambda x:(-x[1], x[0]))
[('f', 5), ('a', 4), ('k', 4), ('g', 3), ('b', 2)]

dic 정렬: lamda식을 이용하여 key, value를 기준으로 정렬

- key 정렬

>>> sorted(dic)
['a', 'b', 'f', 'g', 'k']
>>> sorted(dic.items())
[('a', 4), ('b', 2), ('f', 5), ('g', 3), ('k', 4)]
>>> dict(sorted(dic.items()))
{'a': 4, 'b': 2, 'f': 5, 'g': 3, 'k': 4}

- value 정렬

>>> sorted(dic,key=lambda x:dic[x])
['b', 'g', 'a', 'k', 'f']
>>> sorted(dic.items(), key=lambda x:x[1])
[('b', 2), ('g', 3), ('a', 4), ('k', 4), ('f', 5)]
>>> dict(sorted(dic.items(), key=lambda x:x[1]))
{'b': 2, 'g': 3, 'a': 4, 'k': 4, 'f': 5}

startswith(): 특정 문자열로 시작되는지 확인하는 함수

>>> string = "helloworld"
>>> string.startswith("hello")
True
>>> string.startswith("hello ")
False
>>> string = "Hello, my name is Daniel"

>>> for voc in string.lower().split():
    if voc.startswith("daniel"):
        print("there is Daniel!")

there is Daniel!

lower(), upper(), islower(), isupper(), swapcase(), capitalize()

>>> string
'Hello, my name is Daniel'

>>> string.swapcase()
'hELLO, MY NAME IS dANIEL'

>>> "dddd".capitalize()
'Dddd'

>>> string.upper()
'HELLO, MY NAME IS DANIEL'

>>> string.lower()
'hello, my name is daniel'

>>> string.isupper()
False
>>> string.islower()
False
>>> string.lower().islower()
True

>>> string[10:].upper()
'NAME IS DANIEL'
>>> string = string[0:10] + string[10:].upper()
>>> string
'Hello, my NAME IS DANIEL'
>>> 

set() 연산

• 집합 선언

  >>> s1 = {1,2,3,4,5}
  >>> s2 = {3,4,5,6,7}
  >>> a = [1,2]
  >>> s3 = set(a)
  >>> s3
  {1, 2}

• 합집합 (s.union())

  >>> s4 = s1 | s2
  >>> s4
  {1, 2, 3, 4, 5, 6, 7}

• 교집합 (s.intersection())

  >>> s5 = s1 & s2
  >>> s5
  {3, 4, 5}

• 차집합 (s.difference())

  >>> s2-s1
  {6, 7}

• 집합간의 비교, 교집합이 공집합인지 여부는 s.isdisjoint()

  >>> s1 == s2
  False
  >>> s1 != s2
  True
  >>> s2
  {3, 4, 5, 6, 7}
  >>> s1 = {3}
  >>> s1.isdisjoint(s2)
  False

• add(), update()

  >>> s1.add(9)
  >>> s1
  {1, 2, 3, 4, 5, 9}
  >>> s1.update({1,2,4,5})
  >>> s1
  {1, 2, 3, 4, 5, 9}

• remove(), discard(), pop(): discard()는 제거할 원소가 없어도 에러X

  >>> s1.remove(1)
  >>> s1
  {2, 3, 4, 5, 9}
  >>> s1.discard(10)
  >>> s1.discard(3)
  >>> s1
  {2, 4, 5, 9}

• pop(), clear()

  >>> s1.pop()
  2
  >>> s1
  {4, 5, 9}
  >>> s1.clear()
  >>> s1
  set()

참고문헌

1. https://docs.python.org/3/library/collections.html#collections.Counter
2. https://docs.python.org/3/tutorial/index.html
3. https://blockdmask.tistory.com/451

*sorted() vs list.sort() *

>>> a = [234,35,1]
>>> sorted(a)
[1, 35, 234]
>>> a
[234, 35, 1]
>>> a.sort()
>>> a
[1, 35, 234]

Dictionary (key, value 형식)

>>> a = {}
>>> a[3] = [1,2,3]
>>> a[3][1]
2
>>> a[3][0]
1
>>> a[2] = {1: 'a'}
>>> a
{3: [1, 2, 3], 2: {1: 'a'}}
>>> a[2][1]
'a'
>>> del a[3]
>>> a
{2: {1: 'a'}}

Dictionary 내장함수 (keys(), values(), get(), clear())

>>> test = {1: 123, 2 : 2345, 3: 32435}
>>> test
{1: 123, 2: 2345, 3: 32435}
>>> test.keys()
dict_keys([1, 2, 3])
>>> len(test)
3
>>> for i in test.keys():
    print(i)
1
2
3
>>> a = list(test.keys())
>>> a
[1, 2, 3]

>>> test.values()
dict_values([123, 2345, 32435])
>>> for i in test.values():
    print(i)


123
2345
32435

>>> for i in test.items():
    print(i)
    type(i)
    i[0]
    i[1]


(1, 123)
<class 'tuple'>
1
123
(2, 2345)
<class 'tuple'>
2
2345
(3, 32435)
<class 'tuple'>
3
32435

>>> test.get(1)
123

Dictionary 값 추가 & 삭제 (update(), pop())

>>> test.update(age = 10, height = 100)
>>> test
{'age': 10, 'height': 100} # key 문자열로 인식 

>>> test.update({2018: 2022})

>>> test.pop(2018)
2022
>>> test
{'age': 10, 'height': 100}

*lamda 표현식 *

>>> a = lamda x:x+3
>>> a(1)
4

map(함수, list) and list comprehension

>>> a = [2,4,6,8,10]
>>> c = list(map((lambda x:x//2), a))
>>> c
[1, 2, 3, 4, 5]

>>> list(map(lambda x,y: x+y, a, c))
[140, 188, 236]

>>> c = [x//2 for x in a]
>>> c
[1, 2, 3, 4, 5]

#### 가독성이 떨어지지만 아래와 같이 표현도 가능하다
>>> d = list(map(lambda x:print(x) if x%2==0 else x+1, a))
2
4
6
8
10

• filter()

  >>> c = list(filter(lambda x:x>3, a))
  >>> c
  [4, 6, 8, 10]
  

c = [x*23 for x in a if x>4] c

[138, 184, 230]

- reduce()

from functools import reduce reduce(lambda x,y:xy, a) 3840 reduce(lambda x,y:xy, a, 2) 7680 reduce(lambda x,y:x*y, a, 0) 0

- list comprehension

c = [x2 if x>4 else x5 for x in range(1,10)] c [5, 10, 15, 20, 10, 12, 14, 16, 18]

문자열 공간 채우기 (zfill(), rjust())

• zfill(): 특정 width만큼 0채우기 (단, 이미 width가 채워져 있다면 변화 X)

  >>> a = "0b113"
  >>> a[2:].zfill(5)
  '00113'

• rjust(): 0이아닌 특정 문자 지정 가능

  >>> a[2:].rjust(5, "a")
  'aa113'

  특정 원소 인덱스 찾기 (index(), index())

• index()

  >>> a = ["Kim", "Lee"]
  >>> a.index("Lee")
  1
  

a.index("Lim") Traceback (most recent call last): File "<pyshell#2>", line 1, in a.index("Lim") ValueError: 'Lim' is not in list

- find()

a.find("Kim") Traceback (most recent call last): File "<pyshell#3>", line 1, in a.find("Kim") AttributeError: 'list' object has no attribute 'find'

b = "Kim" b.find("i") 1

b.find("h") -1

> ### Binary 계산

- int(): n진수 -> 10진수

int("1111",2) 15 int("1111",3) 40

- bin(), 논리연산

bin(20) '0b10100'

39 | 22 # OR 연산 55 38 & 33 # AND 연산 32 39 ^ 32 # XOR 연산 7 ~39 # NOT 연산 -40

> ### 순열, 조합 

from itertools import permutations, combinations, product, combinations_with_replacement

data = ['A', 'B', "C"]

result = list(permutations(data, 2)) print(result) # 순열

result1 = list(combinations(data,2)) print(result1) # 조합

result2 = list(product(data, repeat=2)) print(result2) # 중복순열

result3 = list(combinations_with_replacement(data, 2)) print(result3) #중복조합

[('A', 'B'), ('A', 'C'), ('B', 'A'), ('B', 'C'), ('C', 'A'), ('C', 'B')] [('A', 'B'), ('A', 'C'), ('B', 'C')] [('A', 'A'), ('A', 'B'), ('A', 'C'), ('B', 'A'), ('B', 'B'), ('B', 'C'), ('C', 'A'), ('C', 'B'), ('C', 'C')] [('A', 'A'), ('A', 'B'), ('A', 'C'), ('B', 'B'), ('B', 'C'), ('C', 'C')]

> ### 문자열에서 특정 문자 교체

- replace()

a 'hellohelloworld' a = a.replace("hello","stop") a 'stopstopworld' a = a.replace("stop", "",1) a 'stopworld' a = a.replace("stop", "hello") a 'helloworld' a.replace("hello","j").replace("w","k") 'jkorld' a = 'helloworld' a = a[:2].replace("he", "hello") a 'hello'

> ### find(), join(), split()을 이용한 문자열 삽입

myIntro = "{} is {} years old".format(myName, myAge) myIntro = "Daniel is 25 years old" idx = myIntro.find("25") idx 10 myProfile = myIntro[:idx] + "good at soccer and " + myIntro[idx:] myProfile 'Daniel is good at soccer and 25 years old'

stringList = myProfile.split() stringList.insert(5, "and polite ") myProfile = ' '.join(stringList) myProfile 'Daniel is good at soccer and polite and 25 years old'

> ### 리스트 요소 삭제

- del list[ ]

a = [1,2,3,4,5] del a[1] a [1, 3, 4, 5]

- remove()

b = [1,1,1,1,66,88,99] b.remove(1) b [1, 1, 1, 66, 88, 99] for i in b: b.remove(3) print(b)

Traceback (most recent call last): File "<pyshell#129>", line 2, in b.remove(3) ValueError: list.remove(x): x not in list

리스트 내에 제거할 원소가 있어야 한다

for i in b: b.remove(1) print(b)

[1, 1, 66, 88, 99] [1, 66, 88, 99] [66, 88, 99]