Nemotron-Personas-Korea 실습: 불러오기, 기초통계, 직무별로 보기

import os
import sys
from pathlib import Path

# 이 PC에서는 Hugging Face의 Xet 전송이 멈춘다. huggingface_hub를 import하기 전에 꺼 둔다.
os.environ.setdefault("HF_HUB_DISABLE_XET", "1")

# 프로젝트 폴더를 찾는다. 노트북을 어느 폴더에서 열어도 scripts/config.py가 있는 곳까지 올라간다.
ROOT = Path.cwd().resolve()
while not (ROOT / "scripts" / "config.py").exists() and ROOT != ROOT.parent:
    ROOT = ROOT.parent
sys.path.insert(0, str(ROOT / "scripts"))

try:
    import config as C  # 경로, 시드, 색상은 모두 여기(.env)에서 온다
except ModuleNotFoundError:
    # 노트북 파일만 따로 받은 경우다. config.py 없이 같은 기본값으로 돌린다. 데이터는 노트북 옆 data/raw에 받는다
    from types import SimpleNamespace

    ROOT = Path.cwd().resolve()
    _data = Path(os.environ.get("NPK_DATA_DIR", "data/raw"))
    _data = _data if _data.is_absolute() else ROOT / _data
    _data.mkdir(parents=True, exist_ok=True)
    C = SimpleNamespace(
        ROOT=ROOT,
        DATA_DIR=_data,
        REPO_ID=os.environ.get("NPK_REPO_ID", "nvidia/Nemotron-Personas-Korea"),
        SEED=int(os.environ.get("NPK_SEED", "42")),
        FONT=os.environ.get("NPK_FONT", "Malgun Gothic"),
        COLOR_1=os.environ.get("NPK_COLOR_1", "#2a78d6"),
        COLOR_2=os.environ.get("NPK_COLOR_2", "#eb6834"),
        COLOR_SURFACE=os.environ.get("NPK_COLOR_SURFACE", "#fcfcfb"),
        COLOR_GRID=os.environ.get("NPK_COLOR_GRID", "#e1e0d9"),
        parquet_glob=lambda: str(_data / "**" / "*.parquet"),
    )

import matplotlib.pyplot as plt
import polars as pl   # pandas 대신 polars를 쓴다. 100만 행 집계가 빠르고 필요한 컬럼만 읽을 수 있다

# 그림을 창으로 띄우지 않고 셀 바로 아래에 그린다

# 그림의 공통 모양. 색과 글꼴 값은 직접 적지 않고 config.py의 토큰을 가져다 쓴다
plt.rcParams.update({
    "font.family": C.FONT,              # 한글 글꼴
    "axes.unicode_minus": False,        # 한글 글꼴에서 마이너스 기호가 깨지는 것을 막는다
    "figure.facecolor": C.COLOR_SURFACE,
    "axes.facecolor": C.COLOR_SURFACE,
    "axes.edgecolor": C.COLOR_GRID,
    "axes.grid": True,
    "grid.color": C.COLOR_GRID,
    "grid.linewidth": 0.6,
    "axes.spines.top": False,
    "axes.spines.right": False,
    "axes.axisbelow": True,
    "figure.dpi": 110,
})

# 표 출력 설정: 최대 40행, 컬럼은 생략 없이 전부(-1), 문자열은 60자까지 보여 준다
pl.Config(tbl_rows=40, tbl_cols=-1, fmt_str_lengths=60, tbl_width_chars=160, tbl_hide_dataframe_shape=True)

print("프로젝트 폴더:", ROOT)
print("설정 파일    :", "있음" if (ROOT / ".env").exists() else "없음 (config.py의 기본값을 쓴다)")
print("데이터 폴더  :", C.DATA_DIR)

# 데이터 폴더와 그 아래 폴더에서 parquet 파일을 모두 찾는다
files = sorted(C.DATA_DIR.rglob("*.parquet"))

if not files:   # 하나도 없을 때만 내려받는다
    from huggingface_hub import snapshot_download

    snapshot_download(
        repo_id=C.REPO_ID,              # "nvidia/Nemotron-Personas-Korea"
        repo_type="dataset",            # 모델이 아니라 데이터셋 저장소다
        local_dir=str(C.DATA_DIR),      # 캐시 폴더가 아니라 프로젝트의 data/raw에 바로 둔다
        allow_patterns=["*.parquet", "*.md", "*.json"],   # 데이터와 설명 문서만 받는다
        max_workers=int(os.environ.get("NPK_DL_WORKERS", "1")),   # 느린 회선에서는 한 개씩 받는 편이 낫다
    )
    files = sorted(C.DATA_DIR.rglob("*.parquet"))

# 파일 크기를 더해 GB로 바꾼다
size_gb = sum(f.stat().st_size for f in files) / 1024**3
print(f"parquet 파일 {len(files)}개, {size_gb:.2f} GB")

lf = pl.scan_parquet(C.parquet_glob())           # 파일 9개를 하나의 표처럼 다룬다. 아직 읽지는 않는다 (LazyFrame)
schema = lf.collect_schema()                     # 컬럼 이름과 자료형. 파일 머리말만 읽으므로 빠르다
n_rows = lf.select(pl.len()).collect().item()    # 행 수. collect()를 불러야 실제로 계산된다

print(f"{n_rows:,}행 x {len(schema)}컬럼")

# 셀의 마지막 줄에 표를 두면 노트북이 표로 그려 준다
pl.DataFrame({"컬럼": list(schema.names()), "자료형": [str(t) for t in schema.dtypes()]})

# 인구통계 변수. country는 값이 하나뿐이라 뺐다
DEMO_COLS = ["sex", "age", "marital_status", "military_status", "family_type", "housing_type",
             "education_level", "bachelors_field", "occupation", "district", "province"]
# 4절에서 직무로 거른 뒤에 읽을 서술문 컬럼
TEXT_COLS = ["persona", "professional_persona", "skills_and_expertise_list", "career_goals_and_ambitions"]

# 인구통계 변수만 골라 메모리에 올린다. 여기서부터 demo는 일반 DataFrame이다
demo = lf.select(["uuid"] + DEMO_COLS).collect()
print(f"메모리에 올린 크기: {demo.estimated_size('mb'):.0f} MB")
demo.head(5)

# 컬럼마다 한 줄씩: 빈 값이 몇 개인지, 서로 다른 값이 몇 가지인지, 가장 흔한 값이 무엇인지
pl.DataFrame({
    "컬럼": DEMO_COLS,
    "결측": [demo[c].null_count() for c in DEMO_COLS],
    "고유값 수": [demo[c].n_unique() for c in DEMO_COLS],
    "가장 많은 값": [str(demo[c].mode().sort()[0]) for c in DEMO_COLS],   # 최빈값이 여럿이면 정렬해서 첫 번째를 쓴다
})

# 나이의 요약 통계. alias()는 결과 컬럼에 이름을 붙인다
print(demo.select(
    pl.col("age").min().alias("최소"),
    pl.col("age").quantile(0.25).alias("1사분위"),
    pl.col("age").median().alias("중앙값"),
    pl.col("age").mean().round(1).alias("평균"),
    pl.col("age").quantile(0.75).alias("3사분위"),
    pl.col("age").max().alias("최대"),
))

bins = range(15, 105, 5)   # 15세부터 5세 간격으로 구간을 나눈다
fig, ax = plt.subplots(figsize=(9, 3.6))
# 성별마다 선을 하나씩 그린다. 막대를 겹치면 가려지므로 테두리만 그리는 step 모양을 쓴다
for sex, color in zip(sorted(demo["sex"].unique()), [C.COLOR_1, C.COLOR_2]):
    ax.hist(demo.filter(pl.col("sex") == sex)["age"], bins=bins, histtype="step", linewidth=2, color=color, label=sex)
ax.set_xlabel("나이")
ax.set_ylabel("인원")
ax.set_title("성별 나이 분포 (5세 구간)")
ax.legend(frameon=False)
plt.show()

# 학력을 낮은 것부터 높은 것 순으로 적어 둔다. 그림의 축 순서와 '전문대 이상' 계산에 쓴다
EDU_ORDER = ["무학", "초등학교", "중학교", "고등학교", "2~3년제 전문대학", "4년제 대학교", "대학원"]


def share(df, col):
    """값별 인원과 비율(%). 많은 순으로 정렬한다."""
    return (
        df.group_by(col).len()                                  # 값별로 묶어 행 수를 센다 (결과 컬럼 이름은 len)
        .with_columns((pl.col("len") / pl.col("len").sum() * 100).round(1).alias("비율(%)"))   # 전체 합으로 나눠 %로
        .rename({"len": "인원"})
        .sort("인원", descending=True)
    )


for col in ["sex", "marital_status", "education_level", "housing_type"]:
    print(f"[{col}]")
    print(share(demo, col))

fig, axes = plt.subplots(1, 2, figsize=(13, 4.6))   # 그림 두 개를 옆으로 나란히

edu = share(demo, "education_level")
# 학력은 인원순이 아니라 EDU_ORDER 순서로 놓는다. 학력마다 순번(o)을 붙이고 그 순번으로 정렬한다
edu = edu.with_columns(pl.col("education_level").replace_strict({k: i for i, k in enumerate(EDU_ORDER)}, default=-1).alias("o")).sort("o")
axes[0].barh(edu["education_level"], edu["비율(%)"], color=C.COLOR_1)
axes[0].set_title("학력 분포 (%)")

prov = share(demo, "province").sort("인원")   # 가로 막대는 아래에서 위로 쌓이므로 오름차순으로 두면 큰 값이 위에 온다
axes[1].barh(prov["province"], prov["비율(%)"], color=C.COLOR_1)
axes[1].set_title("시도별 분포 (%)")

for ax in axes:
    ax.grid(axis="y", visible=False)   # 가로 막대에서는 세로 눈금선만 남긴다
plt.tight_layout()
plt.show()

FORMER_MARK = "현재 구직중"   # '전직 일반 간호사, 현재 구직중' 같은 값을 가려내는 표지
NO_JOB = "무직"

# 직업 값을 보고 세 가지 상태로 나눈 컬럼(work_status)을 붙인다. when-then-otherwise는 if-elif-else와 같다
demo = demo.with_columns(
    pl.when(pl.col("occupation") == NO_JOB).then(pl.lit("무직"))
    .when(pl.col("occupation").str.contains(FORMER_MARK, literal=True)).then(pl.lit("전직(구직중)"))
    .otherwise(pl.lit("취업자"))
    .alias("work_status")
)
print(share(demo, "work_status"))

employed = demo.filter(pl.col("work_status") == "취업자")   # 4절에서 비교 기준으로 쓴다
print(f"직업 고유값: 전체 {demo['occupation'].n_unique():,}개, 취업자만 {employed['occupation'].n_unique():,}개")

# 취업자 가운데 많은 직업 20개. 여기의 비율(%)은 취업자 안에서의 비율이다
share(employed, "occupation").head(20)

# 직업명에 이 낱말 가운데 하나라도 들어가면 같은 직무로 묶는다. 환경변수로도 줄 수 있다 (쉼표로 구분).
JOB_KEYWORDS = [k.strip() for k in os.environ.get("NPK_JOB_KEYWORDS", "간호").split(",") if k.strip()]

# 걸린 직업명 가운데 빼고 싶은 것. 예: ["간호조무사"]
JOB_EXCLUDE = [k.strip() for k in os.environ.get("NPK_JOB_EXCLUDE", "").split(",") if k.strip()]

# '전직 ○○, 현재 구직중'도 같은 직무로 볼 것인가
INCLUDE_FORMER = os.environ.get("NPK_JOB_INCLUDE_FORMER", "0") == "1"

# 서술문 예시를 몇 명 볼 것인가
N_EXAMPLES = int(os.environ.get("NPK_JOB_N_EXAMPLES", "3"))

print("직무 낱말:", JOB_KEYWORDS, "| 제외:", JOB_EXCLUDE, "| 전직 포함:", INCLUDE_FORMER)

def job_expr(keywords, exclude=(), include_former=False):
    """직업명 조건을 polars 식으로 만든다. 낱말은 정규식이 아니라 글자 그대로 찾는다."""
    occ = pl.col("occupation")
    # 낱말마다 '직업명에 들어 있는가'를 따지고, 하나라도 참이면 참으로 본다 (OR)
    hit = pl.any_horizontal([occ.str.contains(k, literal=True) for k in keywords])
    # 제외 낱말이 들어 있는 직업명은 뺀다. ~는 부정(NOT), &는 그리고(AND)
    for k in exclude:
        hit = hit & ~occ.str.contains(k, literal=True)
    # '전직 ○○, 현재 구직중'을 빼는 것이 기본이다
    if not include_former:
        hit = hit & ~occ.str.contains(FORMER_MARK, literal=True)
    return hit   # 아직 계산된 값이 아니라 조건식이다. filter()에 넣을 때 계산된다


def find_jobs(keywords, exclude=(), include_former=True):
    """낱말에 걸리는 직업명과 인원. 직무를 정하기 전에 무엇이 섞여 들어오는지 확인하는 용도다."""
    if isinstance(keywords, str):   # find_jobs("간호")처럼 낱말 하나만 줘도 되게 한다
        keywords = [keywords]
    return share(demo.filter(job_expr(keywords, exclude, include_former)), "occupation")


# 검색 단계에서는 전직도 함께 보여 준다 (include_former=True가 기본). 무엇이 있는지 알고 나서 뺄지 정한다
find_jobs(JOB_KEYWORDS, JOB_EXCLUDE)

def job_view(keywords, exclude=(), include_former=False, with_text=True):
    """직무 집단의 행을 돌려준다. with_text=True면 서술문 컬럼도 함께 읽는다."""
    if isinstance(keywords, str):
        keywords = [keywords]
    cond = job_expr(keywords, exclude, include_former)
    if not with_text:
        return demo.filter(cond)   # 메모리에 있는 표에서 거르므로 빠르다
    # 서술문이 필요하면 parquet에서 다시 읽는다. 조건에 맞는 행만, 필요한 컬럼만 읽는다
    return lf.filter(cond).select(["uuid"] + DEMO_COLS + TEXT_COLS).collect()


def job_summary(df, label):
    """직무 집단 한 줄 요약."""
    n = df.height   # 행 수
    # 전공은 대학에 가지 않은 사람('해당없음')을 빼고 가장 흔한 것을 찾는다
    top_field = df.filter(pl.col("bachelors_field") != "해당없음")["bachelors_field"].mode()
    return {
        "직무": label,
        "인원": n,
        "전체 대비(%)": round(n / demo.height * 100, 2),
        "취업자 대비(%)": round(n / employed.height * 100, 2),
        "직업명 수": df["occupation"].n_unique(),                    # 이 직무로 묶인 직업명이 몇 가지인가
        "여자(%)": round((df["sex"] == "여자").mean() * 100, 1),      # 참/거짓의 평균이 곧 비율이다
        "나이 중앙값": df["age"].median(),
        "전문대 이상(%)": round(df["education_level"].is_in(EDU_ORDER[4:]).mean() * 100, 1),   # EDU_ORDER[4:] = 전문대, 4년제, 대학원
        "가장 많은 전공": top_field.sort()[0] if len(top_field) else "-",
        "가장 많은 시도": df["province"].mode().sort()[0],
    }


JOB_LABEL = "+".join(JOB_KEYWORDS)   # 표와 그림에 쓸 직무 이름
job = job_view(JOB_KEYWORDS, JOB_EXCLUDE, INCLUDE_FORMER)   # 4절의 나머지 셀은 모두 이 job을 쓴다
if job.height == 0:   # 낱말이 틀렸을 때 뒤에서 엉뚱한 오류가 나지 않도록 여기서 멈춘다
    raise ValueError(f"{JOB_KEYWORDS}에 걸리는 직업이 없다. find_jobs()로 낱말을 확인한다.")

pl.DataFrame([job_summary(job, JOB_LABEL)])

def compare_share(job_df, col, base=None, top=None):
    """한 컬럼의 분포를 직무 집단과 기준 집단에서 나란히 본다."""
    base = employed if base is None else base   # 기준 집단을 따로 주지 않으면 취업자 전체
    a = share(job_df, col).select(col, pl.col("인원").alias("직무 인원"), pl.col("비율(%)").alias("직무(%)"))
    b = share(base, col).select(col, pl.col("비율(%)").alias("취업자(%)"))
    out = (
        a.join(b, on=col, how="full", coalesce=True)   # 한쪽에만 있는 값도 남긴다 (예: 이 직무에 없는 학력)
        .fill_null(0)                                  # 없는 쪽은 0%로 채운다
        .with_columns((pl.col("직무(%)") - pl.col("취업자(%)")).round(1).alias("차이(%p)"))
        .sort("직무(%)", descending=True)
    )
    return out.head(top) if top else out


# (컬럼, 위에서 몇 줄만 볼지). None이면 전부 보여 준다
for col, top in [("sex", None), ("education_level", None), ("bachelors_field", 8), ("marital_status", None), ("province", 8)]:
    print(f"[{col}]")
    print(compare_share(job, col, top=top))

fig, axes = plt.subplots(1, 2, figsize=(13, 4.4))

# 왼쪽: 학력 분포
edu = compare_share(job, "education_level")
edu = edu.with_columns(pl.col("education_level").replace_strict({k: i for i, k in enumerate(EDU_ORDER)}, default=-1).alias("o")).sort("o")
y = range(edu.height)   # 학력마다 세로 위치 0, 1, 2, ...
h = 0.38                # 막대 두께. 두 막대를 위아래로 반 칸씩 비켜 놓는다
axes[0].barh([i + h / 2 for i in y], edu["직무(%)"], height=h, color=C.COLOR_2, label=JOB_LABEL)
axes[0].barh([i - h / 2 for i in y], edu["취업자(%)"], height=h, color=C.COLOR_1, label="취업자 전체")
axes[0].set_yticks(list(y))
axes[0].set_yticklabels(edu["education_level"])
axes[0].set_title("학력 분포 (%)")
axes[0].grid(axis="y", visible=False)
axes[0].legend(frameon=False)

# 오른쪽: 나이 분포 (두 집단의 크기가 다르므로 비율로 그린다)
bins = range(15, 105, 5)
axes[1].hist(employed["age"], bins=bins, density=True, histtype="step", linewidth=2, color=C.COLOR_1, label="취업자 전체")
axes[1].hist(job["age"], bins=bins, density=True, histtype="step", linewidth=2, color=C.COLOR_2, label=JOB_LABEL)
axes[1].set_title("나이 분포 (밀도)")
axes[1].set_xlabel("나이")
axes[1].legend(frameon=False)

plt.tight_layout()
plt.show()

# 직업명별로 묶어 여러 값을 한꺼번에 계산한다. agg() 안의 식 하나가 결과 컬럼 하나가 된다
(
    job.group_by("occupation")
    .agg(
        pl.len().alias("인원"),
        (pl.col("sex") == "여자").mean().mul(100).round(1).alias("여자(%)"),
        pl.col("age").median().alias("나이 중앙값"),
        pl.col("education_level").is_in(EDU_ORDER[4:]).mean().mul(100).round(1).alias("전문대 이상(%)"),
        (pl.col("education_level") == "대학원").mean().mul(100).round(1).alias("대학원(%)"),
    )
    .sort("인원", descending=True)
)

# 직업명 x 학력 교차표 (행 기준 %)
ct = (
    job.group_by("occupation", "education_level").len()   # (직업명, 학력) 조합별 인원
    # over("occupation"): 합계를 직업명 안에서만 구한다. 그래서 각 행의 비율을 더하면 100이 된다
    .with_columns((pl.col("len") / pl.col("len").sum().over("occupation") * 100).round(1).alias("pct"))
    .pivot(on="education_level", index="occupation", values="pct")   # 학력 값을 컬럼으로 펼친다
    .fill_null(0)                                                    # 한 명도 없는 조합은 0
)
# 컬럼을 학력 순서대로 놓는다. 이 직무에 없는 학력은 건너뛴다
ct.select(["occupation"] + [c for c in EDU_ORDER if c in ct.columns])

skills = (
    # "['환자 상태 관찰', '의료 행정']" 같은 문자열에서 작은따옴표로 싸인 부분을 모두 찾아 리스트로 만든다
    job.select(pl.col("skills_and_expertise_list").str.extract_all(r"'([^']+)'").alias("s"))
    .explode("s")                                     # 리스트의 항목 하나가 행 하나가 되게 펼친다
    .with_columns(pl.col("s").str.strip_chars("'"))   # 양끝에 남은 작은따옴표를 뗀다
    .drop_nulls()
)
print(f"기술 항목 {skills.height:,}개, 고유 {skills['s'].n_unique():,}개")

words = (
    skills.select(pl.col("s").str.split(" ").alias("w")).explode("w")   # 항목을 띄어쓰기로 쪼개 낱말 하나가 행 하나
    .filter(pl.col("w").str.len_chars() >= 2)                           # '및', '등' 같은 한 글자는 버린다
    .filter(~pl.col("w").is_in(["관리", "기술", "능력", "활용", "통한", "위한"]))   # 어느 직무에나 나오는 낱말
)
share(words, "w").rename({"w": "낱말", "인원": "횟수"}).head(25)

def show_examples(df, n=3, seed=C.SEED):
    """df에서 n명을 무작위로 뽑아 인구통계 변수 한 줄과 서술문 세 가지를 찍는다."""
    # 집단이 n명보다 작으면 있는 만큼만 뽑는다. to_dicts()는 행마다 {컬럼: 값} 사전을 만든다
    for r in df.sample(n=min(n, df.height), seed=seed).to_dicts():
        print("=" * 100)
        print(f"{r['occupation']} | {r['sex']} {r['age']}세 | {r['education_level']} ({r['bachelors_field']}) | {r['province']} {r['district']}")
        print("-" * 100)
        print("[직업 페르소나]", r["professional_persona"])
        print("[기술 목록]    ", r["skills_and_expertise_list"])
        print("[경력 목표]    ", r["career_goals_and_ambitions"])


show_examples(job, N_EXAMPLES)

# 이 직무에 실제로 있는 학력만 낮은 순서대로 남긴다
edu_in_job = [e for e in EDU_ORDER if e in job["education_level"].unique().to_list()]
# 맨 뒤(가장 높은 학력)와 맨 앞(가장 낮은 학력)에서 한 명씩 읽는다
for level in (edu_in_job[-1], edu_in_job[0]):
    print(f"\n### 학력: {level}")
    show_examples(job.filter(pl.col("education_level") == level), n=1)

# 직무 이름: {"keywords": 찾을 낱말들, "exclude": 뺄 낱말들(없어도 된다)}
JOB_SETS = {
    "간호": {"keywords": ["간호사"]},   # '간호사'로 찾으면 간호조무사는 빠진다
    "소프트웨어 개발": {"keywords": ["소프트웨어 개발", "프로그래머", "웹 개발"]},
    "교사(초중등)": {"keywords": ["초등학교 교사", "중학교 교사", "고등학교 교사", "중·고등학교 교사"]},
    "경리·회계 사무": {"keywords": ["경리 사무원", "회계 사무원"]},
    "건물 청소·경비": {"keywords": ["건물 청소원", "건물 경비원"]},
}

rows = []
for label, spec in JOB_SETS.items():
    # 요약에는 서술문이 필요 없으므로 with_text=False로 빠르게 거른다
    df = job_view(spec["keywords"], spec.get("exclude", ()), include_former=False, with_text=False)
    if df.height:
        rows.append(job_summary(df, label))
    else:
        print(f"'{label}'에 걸리는 직업이 없다. find_jobs({spec['keywords']})로 확인한다.")

pl.DataFrame(rows)