pyarrow TableInvalid - Parquet 스키마 충돌 해결

import pyarrow as pa
import pyarrow.parquet as pq

# file1: user_id int64
pq.write_table(
    pa.table({"user_id": pa.array([1, 2], type=pa.int64()), "event": ["a", "b"]}),
    "part1.parquet",
)

# file2: user_id string
pq.write_table(
    pa.table({"user_id": pa.array(["3", "4"], type=pa.string()), "event": ["c", "d"]}),
    "part2.parquet",
)

# 병합 시도
import pyarrow.dataset as ds

dataset = ds.dataset(["part1.parquet", "part2.parquet"], format="parquet")
# 아래에서 TableInvalid가 발생할 수 있음
table = dataset.to_table()

import pyarrow as pa
import pyarrow.parquet as pq
from pathlib import Path

TARGET_SCHEMA = pa.schema([
    ("user_id", pa.string()),
    ("event", pa.string()),
])

def normalize_table(tbl: pa.Table) -> pa.Table:
    # 누락 컬럼 채우기
    for field in TARGET_SCHEMA:
        if field.name not in tbl.column_names:
            tbl = tbl.append_column(field.name, pa.nulls(tbl.num_rows, type=field.type))

    # 불필요 컬럼 제거 + 컬럼 순서 맞추기
    tbl = tbl.select([f.name for f in TARGET_SCHEMA])

    # 타입 캐스팅
    cols = []
    for f in TARGET_SCHEMA:
        arr = tbl[f.name]
        # 안전 캐스팅: 숫자->문자열은 가능, 문자열->숫자는 실패할 수 있음
        cols.append(arr.cast(f.type))

    return pa.Table.from_arrays(cols, schema=TARGET_SCHEMA)

in_files = ["part1.parquet", "part2.parquet"]
out_dir = Path("normalized")
out_dir.mkdir(exist_ok=True)

for p in in_files:
    t = pq.read_table(p)
    t2 = normalize_table(t)
    pq.write_table(t2, out_dir / Path(p).name)

import pyarrow as pa
import pyarrow.compute as pc

TARGET_TS = pa.timestamp("us", tz="UTC")

def normalize_ts(arr: pa.ChunkedArray) -> pa.ChunkedArray:
    # 1) timezone 없는 timestamp면 UTC로 가정(정책에 따라 변경)
    t = arr.type
    out = arr

    if pa.types.is_timestamp(t) and t.tz is None:
        # tz 부여는 cast로 직접 되지 않는 케이스가 있어, 문자열 변환을 쓰면 비용이 큼
        # 여기서는 단순 예시로 cast 시도
        out = out.cast(pa.timestamp(t.unit, tz="UTC"))

    # 2) 단위를 us로 통일
    if pa.types.is_timestamp(out.type) and out.type.unit != "us":
        out = out.cast(pa.timestamp("us", tz=out.type.tz))

    # 3) 최종 타입 강제
    out = out.cast(TARGET_TS)
    return out

import pyarrow as pa

def undict(arr: pa.ChunkedArray) -> pa.ChunkedArray:
    if pa.types.is_dictionary(arr.type):
        return arr.cast(arr.type.value_type)
    return arr

import pyarrow.parquet as pq
from collections import defaultdict

files = ["part1.parquet", "part2.parquet"]

schemas = {}
for f in files:
    pf = pq.ParquetFile(f)
    schemas[f] = pf.schema_arrow

# 컬럼별 타입 분포 확인
col_types = defaultdict(set)
for f, s in schemas.items():
    for field in s:
        col_types[field.name].add(str(field.type))

for col, types in sorted(col_types.items()):
    if len(types) > 1:
        print(col, types)

from pathlib import Path
import pyarrow as pa
import pyarrow.parquet as pq
import pyarrow.dataset as ds

TARGET_SCHEMA = pa.schema([
    ("user_id", pa.string()),
    ("event", pa.string()),
    ("event_ts", pa.timestamp("us", tz="UTC")),
])

def normalize(tbl: pa.Table) -> pa.Table:
    # 누락 컬럼 채우기
    for f in TARGET_SCHEMA:
        if f.name not in tbl.column_names:
            tbl = tbl.append_column(f.name, pa.nulls(tbl.num_rows, type=f.type))

    # 컬럼 정렬 및 불필요 컬럼 제거
    tbl = tbl.select([f.name for f in TARGET_SCHEMA])

    arrays = []
    for f in TARGET_SCHEMA:
        col = tbl[f.name]
        # dictionary 제거
        if pa.types.is_dictionary(col.type):
            col = col.cast(col.type.value_type)
        # timestamp 표준화
        if pa.types.is_timestamp(f.type):
            col = col.cast(f.type)
        else:
            col = col.cast(f.type)
        arrays.append(col)

    return pa.Table.from_arrays(arrays, schema=TARGET_SCHEMA)

src_dir = Path("raw_parquet")
dst_dir = Path("curated_parquet")
dst_dir.mkdir(exist_ok=True)

for f in src_dir.glob("*.parquet"):
    t = pq.read_table(f)
    t2 = normalize(t)
    pq.write_table(t2, dst_dir / f.name)

dataset = ds.dataset(str(dst_dir), format="parquet")
merged = dataset.to_table()
print(merged.schema)
print(merged.num_rows)