Python으로 데이터 직렬화2

%cd python-serialize

HTTP 메시지 페이로드 직렬화

모든 HTTP 메시지는 세 부분으로 구성됩니다.

1. 요청 줄 또는 응답 상태 줄
2. Head
3. Body

Flask를 사용하여 REST API 구축

In [ ]:

# flask-rest-api/main.py

from dataclasses import dataclass
from datetime import datetime
from uuid import UUID, uuid4

from flask import Flask, jsonify, request

app = Flask(__name__)

@dataclass
class User:
    id: UUID
    name: str
    created_at: datetime

    @classmethod
    def create(cls, name):
        return cls(uuid4(), name, datetime.now())

users = [
    User.create("Alice"),
    User.create("Bob"),
]

@app.route("/users", methods=["GET", "POST"])
def view_users():
    if request.method == "GET":
        return users
    elif request.method == "POST":
        if request.is_json:
            payload = request.get_json()
            user = User.create(payload["name"])
            users.append(user)
            return jsonify(user), 201

• 9~17행은 User 모델을 세 가지 속성이 있는 Python 데이터 클래스로 정의합니다. 명명된 생성자, .create()는 무작위 UUID 및 현재 날짜와 시간을 사용하여 이들 속성중에 2가지를 설정합니다.
• 19~22행 Alice와 Bob이라는 두 명의 샘플 사용자가 있는 사용자 컬렉션을 지정합니다.
• 24~33행은 HTTP GET 및 HTTP POST 방법에 응답하는 유일한 REST 엔드포인트, /users를 정의합니다.
• 27행은 Flask가 JSON 배열로 직렬화하는 모든 사용자 목록을 반환합니다.
• 30~32행 페이로드에서 역직렬화된 JSON을 기반으로 새 사용자를 생성하고 이를 모든 사용자 컬렉션에 추가합니다.
• 33행은 새로 생성된 사용자와 HTTP 201 상태 코드로 구성하는 튜플을 반환합니다. Flask가 데이터 클래스의 인스턴스를 직렬화하는 데 도움을 주기 위해 반환할 사용자 개체에 대해 jsonify()을 호출합니다.

성능을 위해 FastAPI 활용

Flask와 마찬가지로 뷰 함수에서 사전이나 목록을 반환하여 JSON 대응 항목으로 자동 직렬화되도록 할 수 있습니다. 이 외에도 FastAPI는 Pydantic 모델과 JSON 형식 페이로드 간의 변환을 지원하므로 Django에서처럼 사용자 정의 직렬 변환기를 작성할 필요가 없습니다.

In [ ]:

# fastapi-rest-api/main.py

from datetime import datetime
from uuid import UUID, uuid4

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

class UserIn(BaseModel):
    name: str

class UserOut(UserIn):
    id: UUID = Field(default_factory=uuid4)
    created_at: datetime = Field(default_factory=datetime.now)

users = [
    UserOut(name="Alice"),
    UserOut(name="Bob"),
]

@app.get("/users")
async def get_users():
    return users

@app.post("/users", status_code=201)
async def create_user(user_in: UserIn):
    user_out = UserOut(name=user_in.name)
    users.append(user_out)
    return user_out
    

• 9행과 10행은 수신 HTTP POST 요청에서 페이로드로 수신되는 .name 속성으로 구성된 사용자 모델을 정의합니다.
• 12~14행은 두 가지를 추가하여 상속을 통해 전자를 확장하는 또 다른 사용자 모델을 정의합니다. 추가 속성은 생성 시 자동으로 초기화됩니다. 이것이 고객에게 다시 보내질 내용입니다.
• 16~19행은 Alice와 Bob이라는 두 명의 샘플 사용자로 채워진 사용자 컬렉션을 지정합니다.
• 21~23행은 사용자 목록을 반환하는 HTTP GET 요청에 의해 트리거되는 API 엔드포인트를 정의합니다.
• 25~29행은 컬렉션에 새 사용자를 추가하고 이를 클라이언트에 반환하여 HTTP POST 요청에 응답하는 또 다른 API 엔드포인트를 정의합니다.

계층적 데이터 직렬화

텍스트: XML, YAML 및 JSON

# XML <program> <metadata> <author>John Doe</author> <goals> <goal>health improvement</goal> <goal>fat loss</goal> </goals> </metadata> <exercises> <exercise id="plank"> <muscles> <muscle>abs</muscle> <muscle>core</muscle> <muscle>shoulders</muscle> </muscles> </exercise> <exercise id="push-ups"> <muscles> <muscle>chest</muscle> <muscle>biceps</muscle> <muscle>triceps</muscle> </muscles> </exercise> </exercises> <days> <day id="rest-day"> <type>rest</type> </day> <day id="workout-1"> <type>workout</type> <routine> <segment> <type ref="plank"/> <duration seconds="60"/> </segment> <segment> <type>rest</type> <duration seconds="10"/> </segment> <segment> <type ref="push-ups"/> <duration seconds="60"/> </segment> </routine> </day> </days> <schedule> <day ref="workout-1"/> <day ref="rest-day"/> <day ref="rest-day"/> <day ref="workout-1"/> <day ref="rest-day"/> </schedule> </program>

표준 라이브러리나 여러 타사 패키지를 사용하여 Python에서 이러한 XML 문서를 역직렬화하는 여러 가지 옵션이 있습니다.

Python과 함께 제공되는 XML 파서는 다음과 같습니다.

• xml.dom.minidom
• xml.sax
• xml.dom.pulldom
• xml.etree.ElementTree

또한 다음을 포함하여 XML 문서를 구문 분석하거나 덤프하기 위해 외부 Python 라이브러리를 설치할 수 있습니다.

• untangle
• xmltodict
• lxml
• lxml.objectify
• BeautifulSoup

# YAML program: metadata: author: John Doe goals: - health improvement - fat loss exercises: - &plank muscles: - abs - core - shoulders - &pushups muscles: - chest - biceps - triceps days: - &restday type: rest - &workout1 type: workout segments: - type: *plank duration: seconds: 60 - type: rest duration: seconds: 10 - type: *pushups duration: seconds: 60 schedule: - day: *workout1 - day: *restday - day: *restday - day: *workout1 - day: *restday

In [ ]:

import yaml
with open("training.yaml", encoding="utf-8") as file:
    document = yaml.safe_load(file)

document["program"]["schedule"][0]["day"]

# JSON { "program": { "metadata": { "author": "John Doe", "goals": ["health improvement", "fat loss"] }, "exercises": { "plank": {"muscles": ["abs", "core", "shoulders"]}, "push-ups": {"muscles": ["chest", "biceps", "triceps"]} }, "days": { "rest-day": {"type": "rest"}, "workout-1": { "type": "workout", "segments": [ {"type": "@plank", "seconds": 60}, {"type": "rest", "seconds": 10}, {"type": "@push-ups", "seconds": 60} ] } }, "schedule": [ "@workout-1", "@rest-day", "@rest-day", "@workout-1", "@rest-day" ] } }

바이너리: BSON

In [ ]:

import json
import bson

with open("training.json", encoding="utf-8") as file:
    document = json.load(file)
    bson.encode(document)

표 형식 데이터 직렬화

pandas, Excel 또는 SQL과 같은 도구를 사용하여 데이터 세트를 탐색할 때 표 형식 데이터로 작업하는 경우가 많습니다. 이 데이터 형태는 서로 다른 시스템 간에 데이터 모델을 저장하고 공유하는 데 적합합니다. 이것이 바로 데이터베이스가 CSV와 같은 형식을 사용하여 테이블을 내보내거나 가져올 수 있는 이유입니다. 표 형식 데이터는 대규모 데이터 세트를 효율적으로 처리하는 역할을 하기 때문에 빅 데이터에서도 자주 발견됩니다.

행 지향열 지향

저장 공간	서로 다른 유형의 열이 함께 저장되므로 더 높습니다.	열 단위 압축으로 컴팩트함
글쓰기	특히 전체 레코드를 쓰거나 업데이트할 때 빠릅니다.	단일 레코드가 여러 열에 분산되어 있기 때문에 속도가 느립니다.
독서	단일 레코드의 모든 열을 읽는 데 최적화되었습니다.	많은 레코드의 특정 열을 읽는 데 최적화됨
쿼리 중	항상 모든 열을 스캔하므로 속도가 느려집니다.	한 번에 몇 개의 열을 쿼리하는 경우에도 빠릅니다.
예	CSV, 아파치 아브로	아파치 쪽모이 세공 마루

텍스트 스프레드시트: CSV

In [ ]:

# csv-deemo/models.py

from datetime import datetime
from enum import StrEnum
from typing import NamedTuple

class Language(StrEnum):
    DE = "de"
    EN = "en"
    ES = "es"
    FR = "fr"
    IT = "it"

class User(NamedTuple):
    id: int
    name: str
    email: str
    language: Language
    registered_at: datetime

사용자에게는 내장 유형, 표준 라이브러리 클래스 및 사용자 정의 클래스가 혼합된 5개의 속성이 있습니다. 그 중 하나는 datetime 인스턴스이고 다른 하나는 바로 위에서 정의한 언어 코드의 사용자 정의 열거입니다.

In [ ]:

# csv-demo/models.py

import random
from datetime import datetime
from enum import StrEnum
from typing import NamedTuple

from faker import Faker

# ...

class User(NamedTuple):
    id: int
    name: str
    email: str
    language: Language
    registered_at: datetime

    @classmethod
    def fake(cls):
        language = random.choice(list(Language))
        generator = Faker(language)
        return cls(
            generator.pyint(),
            generator.name(),
            generator.email(),
            language,
            generator.date_time_this_year(),
        )

클래스에 .fake()이라는 명명된 생성자를 추가했습니다. 이 생성자는 가짜 속성 값이 채워진 새 인스턴스를 생성하고 반환합니다. 여기에서 초기화할 언어를 무작위로 선택합니다. 생성기를 사용하여 해당 값을 생성합니다.

In [ ]:

import csv

with open("users.csv", mode="w", encoding="utf-8", newline="") as file:
    writer = csv.writer(file)
    writer.writerows(users)

이 코드를 실행하면 users.csv이라는 파일이 현재 작업 디렉토리에 다음과 같은 내용으로 나타나는 것을 볼 수 있습니다.

In [ ]:

with open("users.csv", mode="r", encoding="utf-8", newline="") as file:
    reader = csv.reader(file)
    next(reader)

In [ ]:

with open("users.csv", mode="r", encoding="utf-8", newline="") as file:
    reader = csv.DictReader(file, fieldnames=User._fields)
    next(reader)

바이너리 DataFrame: Parquet

Apache Parquet은 표 형식 데이터를 표시하는 데 널리 사용되는 또 다른 직렬화 형식입니다. 바이너리, 스키마 기반 및 열 형식이므로 CSV 파일에 비해 엄청나게 빠르고 공간 효율적입니다.

Python에서 Parquet 파일을 사용하려면 필요한 라이브러리를 설치해야 합니다. 예를 들어, pandas를 사용하는 경우 테이블에는 두 가지 옵션이 있습니다.

1. pyarrow
2. fastparquet

pandas에는 Parquet 파일과 인터페이스하기 위한 고급 기능이 제공되지만 기본적으로 기본 구현이 부족합니다. 필요한 논리를 제공하는 선택적 종속성 중 하나입니다. 두 종속성이 모두 설치되어 있으면 pandas는 pyarrow의 구현을 선호하지만 DataFrameengine을 읽거나 쓸 때 매개변수를 지정하여 이를 재정의할 수 있습니다.

import pandas as pd df1 = pd.read_parquet("/path/to/file") df2 = pd.read_parquet("/path/to/file", engine="fastparquet") df2.to_parquet("/path/to/another_file", engine="pyarrow")

이러한 라이브러리를 pandas와 함께 사용하는 것 외에도 자체적으로 사용할 수도 있습니다.

import pyarrow.parquet as pq table = pq.read_table("/path/to/file") df1 = table.to_pandas() import fastparquet pf = fastparquet.ParquetFile("/path/to/file") df2 = pf.to_pandas()

pyarrow가 노출하는 Parquet의 가장 강력한 기능 중 하나는 조건자 푸시다운 필터링이라고도 알려진 행 필터링입니다. 이는 추가 처리를 위해 메모리로 읽기 전에 대규모 데이터 세트에서 행의 작은 하위 집합만 요청할 수 있으므로 성능, 메모리 사용 및 네트워크 대역폭에 상당한 영향을 미칩니다.

import pandas as pd df = pd.read_parquet( "users.parquet", filters=[("language", "=", "fr")], engine="pyarrow" ) df.head()

Parquet을 포함한 모든 열 형식 데이터 형식의 또 다른 뛰어난 기능은 열 가지치기 또는 프로젝션입니다.

df = pd.read_parquet( "users.parquet", filters=[("language", "=", "fr")], columns=["id", "name"], engine="pyarrow" ) df.head()

스키마 기반 데이터 직렬화

빅데이터: 아파치 아브로

• 효율성: 바이너리 형식이므로 데이터를 유지해야 하거나 유선으로 전송해야 하는지 여부에 관계없이 컴팩트하고 빠르게 처리할 수 있습니다.
• 데이터 일관성: 항상 데이터 직렬화 및 역직렬화를 위한 스키마를 사용하므로 데이터 손상이나 잘못된 해석의 위험이 줄어듭니다.
• 자체 설명: 모든 메시지에는 해당 스키마가 포함되어 있으므로 소비자는 페이로드를 올바르게 해석하기 위해 이를 미리 알 필요가 없습니다. - 동시에 스키마 없는 메시지를 처리하는 동안 선택적으로 스키마를 별도로 수신하여 성능을 더욱 향상시킬 수 있습니다.
• 하위 호환성: 이 형식은 스키마 발전을 탁월하게 지원하므로 시간이 지남에 따라 데이터 구조를 손상시키지 않고 안전하고 원활하게 변경할 수 있습니다.
• 언어 독립성: 이 형식은 다양한 프로그래밍 언어를 지원하므로 분산형 다중 언어 컴퓨팅 환경에 적합합니다.

Avro는 다음과 같은 고유한 기능을 제공하여 차별화합니다.

• 동적 언어 지원: 원하는 경우 특정 스키마에 맞는 직렬화 코드를 미리 생성하지 않고도 런타임에 Avro 메시지를 사용할 수 있습니다. 이는 선택적 최적화 단계이며 Java와 같이 컴파일되고 정적으로 유형이 지정된 언어에 적합합니다. 따라서 Avro의 공식 Python 바인딩은 스키마에서 코드 생성을 지원하지 않습니다.
• 기호 필드 이름: Avro 스키마에는 기호 필드 이름이 포함되어 있지만 일부 다른 형식에서는 해당 필드에 숫자 식별자를 수동으로 할당해야 합니다. 스키마의 진화된 버전 간의 차이점을 해결하려고 할 때 숫자 작업이 방해가 될 수 있습니다.

다음은 CSV 형식으로 작업할 때 이전에 정의한 User 클래스에 해당하는 Avro 스키마입니다.

# avro-demo/user.avsc { "name": "User", "type": "record", "fields": [ {"name": "id", "type": "long"}, {"name": "name", "type": "string"}, {"name": "email", "type": "string"}, { "name": "language", "type": { "name": "Language", "type": "enum", "symbols": ["de", "en", "es", "fr", "it"] } }, { "name": "registered_at", "type": {"type": "long", "logicalType": "timestamp-millis"} } ] }

이 JSON 문서는 64비트 정수 식별자와 두 개의 문자열 필드, 사용자 정의 언어 코드 열거 및 타임스탬프로 구성된 Avro 레코드를 정의합니다

from fastavro.schema import load_schema from fastavro import writer from models import User users = [User.fake() for _ in range(5)] with open("users.avro", mode="wb") as file: schema = load_schema("user.avsc") writer(file, schema, [user._asdict() for user in users])from fastavro import reader from models import User with open("users.avro", mode="rb") as file: for record in reader(file): print(User(**record))

타임스탬프 필드는 UTC 시간대를 사용하는 Python datetime 인스턴스로 올바르게 변환되지만 언어 열거는 문자열로 유지됩니다. 이를 원하지 않는 경우 language 키를 적절하게 재정의하여 역직렬화를 사용자 정의할 수 있습니다.

from fastavro import reader from models import Language, User with open("users.avro", mode="rb") as file: for record in reader(file): print(User(**record | { "language": Language(record["language"]) }))

여기에서는 파이프 연산자(|)를 사용하여 두 사전 결합을 수행합니다. 변환된 Language 인스턴스가 있는 오른쪽 항목은 왼쪽 사전의 해당 키-값 쌍을 대체합니다.

마이크로서비스: 프로토콜 버퍼

프로토콜 버퍼 또는 줄여서 Protobuf는 행 기반의 또 다른 바이너리 스키마입니다. 구조화된 데이터를 위한 언어 중립적인 데이터 직렬화 형식입니다. Google이 2008년에 공개 소스로 만들면서 인기를 얻었습니다. 오늘날 많은 조직에서는 서로 다른 기술 스택을 기반으로 하는 다수의 이기종 마이크로서비스 간에 데이터를 효율적으로 전송하기 위해 프로토콜 버퍼를 사용합니다.

Python을 컴파일러의 출력으로 선택하면 생성된 코드는 PyPI에서 찾을 수 있는 타사 protobuf 패키지에 따라 달라집니다. 하지만 먼저 C++로 구현되는 프로토콜 버퍼 스키마 컴파일러(protoc)를 설치해야 합니다. 컴퓨터에서 작동하게 하는 세 가지 옵션이 있습니다.

1. 로컬 컴퓨터에서 C++ 소스 코드 컴파일
2. 플랫폼에 맞는 사전 빌드된 바이너리 릴리스 다운로드
3. 운영 체제에 해당하는 패키지 설치

# shell sudo apt install protobuf-compiler# users.proto syntax = "proto3"; package com.realpython; import "google/protobuf/timestamp.proto"; enum Language { DE = 0; EN = 1; ES = 2; FR = 3; IT = 4; } message User { int64 id = 1; string name = 2; string email = 3; Language language = 4; google.protobuf.Timestamp registered_at = 5; } message Users { repeated User users = 1; }

• 1행은 파일의 나머지 부분이 프로토콜 버퍼 구문의 버전 3을 사용하여 해석되어야 함을 나타냅니다.
• 3행은 정의에 대한 선택적 네임스페이스를 설정하며 이는 Java와 같은 언어에 필요할 수 있습니다.
• 5행은 프로토콜 버퍼가 제공하는 Timestamp 유형을 가져옵니다.
• 7~13행 사용자 정의 Language 열거를 지정합니다. 유형 범위를 지정하는 방법에 따라 예를 들어 User 메시지 유형 아래에 이 열거형을 중첩할 수 있습니다.
• 15~21행 메시지 유형 이름이 지정된 User을 정의합니다. 다섯 개의 필드가 있습니다. 각 필드에는 인덱스가 0인 열거형 멤버와 달리 1부터 시작해야 하는 고유한 숫자 식별자와 유형이 있습니다. 필드 번호 매기기에 대한 권장 사례는 공식 문서를 읽어보세요!
• 23~25행은 되풀이 하는 Users의 시퀀스를 포함하는 또 다른 메시지 User 유형을 정의합니다.

# shell protoc --python_out=. --pyi_out=. users.proto

--python_out 및 --pyi_out 옵션은 모두 생성된 파일을 저장해야 하는 대상 디렉터리 경로를 나타냅니다. Unix 기반 운영 체제에서 점(.)은 현재 작업 디렉터리를 나타냅니다.

위 명령을 실행하면 두 개의 새로운 파일이 나타납니다.

1. users_pb2.py
2. users_pb2.pyi

새 동적으로 합성된 클래스를 사용하여 프로토콜 버퍼로 사용자를 직렬화하는 방법은 다음과 같습니다.

In [ ]:

from users_pb2 import Language as LanguageDAO
from users_pb2 import User as UserDAO
from users_pb2 import Users as UsersDAO

from models import User
users = [User.fake() for _ in range(5)]

users_dao = UsersDAO()
for user in users:
    user_dao = UserDAO()
    user_dao.id = user.id
    user_dao.name = user.name
    user_dao.email = user.email
    user_dao.language = LanguageDAO.Value(user.language.name)
    user_dao.registered_at.FromDatetime(user.registered_at)
    users_dao.users.append(user_dao)

buffer = users_dao.SerializeToString()
print(buffer)

가져오는 동안 생성된 클래스의 이름을 User에서 UserDAO로 변경합니다.

In [ ]:

from models import Language

users_dao = UsersDAO()
users_dao.ParseFromString(buffer)

In [ ]:

users = [
    User(
        id=user_dao.id,
        name=user_dao.name,
        email=user_dao.email,
        language=list(Language)[user_dao.language],
        registered_at=user_dao.registered_at.ToDatetime()
    )
    for user_dao in users_dao.users
]

users[0]

In [ ]:

출처 : https://realpython.com/python-serialize-data