Spark를 사용한 Config-Driven 데이터 표준화 프레임워크 (번역)

Medium 의 Config-Driven Data Standardization Framework using Spark 를 번역한 글입니다.

 

Spark를 사용한 Config-Driven 데이터 표준화 프레임워크

 

https://medium.com/@pallavisinha12/config-driven-data-standardization-framework-using-spark-12aa7c52fae1

오늘날 조직은 다양한 원천 데이터 소스에서 방대한 양의 원시 데이터를 수집합니다. 이 데이터는 종종 일관성이 없고, 구조가 불완전하며 분석과 의사 결정을 위해서 상당한 전처리를 필요로 합니다. 이러한 문제를 해결하기 위해서는 강력한 데이터 표준화 프로세스가 필수적입니다.

먼저 데이터 표준화의 중요성을 간단히 설명하고, 그 후 설정 기반 접근 방식의 장점을 살펴보며, 마지막으로 이 프레임워크의 단계별 개발 과정을 다룹니다. 또한 이 프레임워크를 사용자의 사용 사례에 어떻게 확장할 수 있으며, 어떤 유용한 점이 있는지도 살펴보겠습니다.

 

데이터 표준화란?

데이터 표준화는 다양한 소스에서 수집된 데이터를 공통된 형식으로 변환하는 과정입니다. 여기에는 명명 규칙을 일치시키고, 데이터 유형을 정렬하며, 메타데이터를 추가하는 등의 작업이 포함되어 데이터 간 일관성을 보장합니다.

 

원시 데이터를 표준화해야 하는 이유

1. 데이터 제품 간 일관성: 표준화는 서로 다른 소스에서 수집된 데이터가 동일한 구조를 따르도록 하여 통합 및 분석이 용이하게 만듭니다.
2. 오류 감소: 일관된 명명 규칙, 올바른 데이터 유형 및 변환을 적용함으로써 데이터 처리 및 분석 과정에서 발생할 수 있는 오류를 크게 줄일 수 있습니다.
3. 데이터 활용성 향상: 표준화된 데이터는 다양한 분석 및 보고 목적으로 더 쉽게 접근하고 활용할 수 있으며, 더 나은 통찰력과 비즈니스 결정을 이끌어냅니다.
4. 확장성: 데이터 양이 증가함에 따라 표준화된 데이터 제품은 새로운 데이터 소스와 컬럼을 추가할 때 더 쉽게 관리하고 확장할 수 있습니다.

 

설정(config) 기반 데이터 표준화 프레임워크

이 프레임워크의 목표는 각 데이터 제품에 대해 맞춤형 수집 및 변환 스크립트를 작성하는 대신, 설정 파일을 사용하여 규칙과 매핑을 정의하는 방식을 사용합니다. 이를 통해 표준화 로직이 애플리케이션 코드에서 분리되며, 시스템의 유연성과 유지 관리성을 향상시킵니다.

이 튜토리얼에서는 설정 파일 형식으로 JSON을 사용하고, 코드는 PySpark와 Spark SQL을 활용할 것입니다. 데이터 제품은 Delta 형식으로 저장됩니다.

설정 기반 접근 방식의 주요 장점
1. 유연성: 애플리케이션 코드를 변경하지 않고도 표준화 규칙을 쉽게 수정할 수 있습니다.
2. 확장성: Spark를 사용하여 대규모 데이터 세트를 효율적으로 처리할 수 있습니다.
3. 유지 보수성: 변환 로직이 설정 파일에 중앙 집중화되어 있어 유지 관리와 업데이트가 용이합니다.

우리가 구축할 기능은 다음과 같습니다:
- 컬럼 이름 업데이트
- 데이터 변환
- 컬럼 데이터 유형 업데이트
- 컬럼 설명 메타데이터 업데이트
- 새로운 컬럼 추가

시작하기 전에, 데이터 표준화는 이 게시글에서  다루는 것 이상의 활동을 포괄할 수 있음을 언급하고 싶습니다. 데이터 품질 검증 추가, 추가 메타데이터 통합, 데이터 제품 버전 관리 등 데이터 무결성과 사용성을 높이는 다른 프로세스가 포함될 수 있습니다.

 

환경 설정하기 

우리는 코드 작성에 Databricks Community Edition을 사용할 것이며, 데이터 저장을 위해 Databricks File System(DBFS)을 사용할 것입니다. 그러나 프레임워크는 경로를 인수로 받기 때문에 S3 버킷, ADLS 등 다른 곳에 파일을 저장해도 괜찮습니다.

설정 파일 구조 설명

이 데모에서 사용할 설정 파일 구조는 다음과 같습니다:

설정 파일 구조
- `data_product_name`: 표준화 후 할당할 데이터 제품(DP)의 이름
- `raw_data_product_name`: 원본 데이터 제품의 이름
- `schema`:
  - `source_columns`: (원본 데이터 제품에서 직접 가져온 컬럼)
    - `raw_name`: 원본 데이터 제품의 컬럼 이름
    - `standardized_name`: 원본 컬럼의 표준화된 이름
    - `data_type`: 변환할 컬럼 데이터 유형
    - `sql_transformation`: Spark SQL로 작성된 변환 규칙
  - `new_columns`: (다른 데이터 제품과 조인을 통해 생성된 컬럼)
    - `name`: 생성될 새로운 컬럼 이름
    - `data_type`: 변환할 컬럼의 데이터 유형
    - `sql_transformation`: Spark SQL로 작성된 변환 규칙
  - `metadata` (모든 컬럼이 추가된 후 할당할 메타데이터)
    - `column_descriptions`: 컬럼 설명

 

원본 데이터 제품 예시: supplier. 아래는 표준화할 원본 데이터 제품 supplier입니다.

supplier (Raw data product)

표준화된 데이터 제품 예시: Product. 아래는 Product라는 다른 표준화된 데이터 제품입니다. 새로운 컬럼을 추가하기 위해 이를 사용할 것입니다.

Product (Other Standardized Data Product)

supplier의 JSON 설정 파일

{
    "data_product_name" : "Product_Supplier",
    "raw_data_product_name" : "supplier",
    "schema" : {
        "source_columns" : [
            {
                "raw_name" : "sup_id",
                "standardized_name" : "Supplier_ID",
                "data_type" : "string",
                "sql_transformation" : "CONCAT('SUP', '-' , sup_id)"
            },
            {
                "raw_name" : "name",
                "standardized_name" : "Supplier_Name",
                "data_type" : "string",
                "sql_transformation" : ""
            },
            {
                "raw_name" : "price",
                "standardized_name" : "Purchase_Price",
                "data_type" : "int",
                "sql_transformation" : ""
            },
            {
                "raw_name" : "prod_name",
                "standardized_name" : "Product_Name",
                "data_type" : "string",
                "sql_transformation" : ""
            },
            {
                "raw_name" : "quantity",
                "standardized_name" : "Purchase_Quantity",
                "data_type" : "int",
                "sql_transformation" : ""
            },
            {
                "raw_name" : "",
                "standardized_name" : "Total_Cost",
                "data_type" : "int",
                "sql_transformation" : "price * quantity"
            }
        ],
        "new_columns" : [
            {
                "name" : "Product_ID",
                "data_type" : "string",
                "sql_transformation" : "MERGE INTO delta.`{temp_std_dp_path}` dest USING delta.`dbfs:/FileStore/project/Product` src ON dest.Product_Name = src.Product_Name WHEN MATCHED THEN UPDATE SET dest.Product_ID = src.Product_ID"
            }
        ]
    },
    "column_sequence_order" : [
        "Supplier_ID", "Supplier_Name", "Product_ID", "Product_Name", "Purchase_Price", "Purchase_Quantity", "Total_Cost"
    ],
    "metadata" : {
        "column_descriptions" : {
            "Supplier_ID" : "제품의 공급업체에 대한 고유 식별자",
            "Supplier_Name" : "공급업체의 이름",
            "Purchase_Price" : "공급업체가 제품을 판매하는 가격",
            "Product_Name" : "제품의 이름",
            "Purchase_Quantity" : "공급업체가 보유한 제품의 수량",
            "Total_Cost" : "주어진 구매 가격으로 특정 수량의 제품을 구매하는 데 소요된 총 금액",
            "Product_ID" : "제품에 대한 고유 식별자"
        }
    }
}

원본 데이터 제품(supplier)과 Product 데이터 제품을 Delta 형식으로 저장하고 JSON 설정 파일을 적절한 경로에 업로드합니다.

우리는 풀 로드 프로세스(풀 로드: 데이터를 초기화한 후 재로드) full load process (truncate-load)를 따를 것입니다. 따라서 모든 단계는 임시 또는 스테이징 영역에서 수행되며, 이후 실제 표준화된 데이터 제품 경로에 덮어쓰게 됩니다. 

 

프레임워크 개발

Config Reader의 인터페이스를 정의합니다. Config Reader 클래스는 설정 파일 구조가 다를 경우에도 사용할 수 있도록 구현될 것입니다. 예를 들어, 이 튜토리얼에서는 JSON 설정을 사용하지만, YAML 설정을 사용하려면 해당 형식에 맞춰 새로운 Config Reader 클래스를 생성하면 됩니다.

class ConfigReaderContract(ABC):
    @abstractmethod
    def read_source_columns_schema(self) -> spark.DataFrame:
        pass
    @abstractmethod
    def read_new_columns_schema(self) -> spark.DataFrame:
        pass
    @abstractmethod
    def read_column_descriptions_metadata(self) -> dict:
        pass
    @abstractmethod
    def read_column_sequence_order(self) -> list[str]:
        pass

ConfigReader 클래스 구현

class ConfigReader(ConfigReaderContract):
    def __init__(self, config_path):
        self.config_df = spark.read.option("multiLine", True).json(config_path)

    def read_source_columns_schema(self):
        exploded_df = self.config_df.select(explode(self.config_df["schema"].source_columns).alias("source_columns"))
        source_columns_schema_df = exploded_df.selectExpr(
            "source_columns.raw_name as raw_name",
            "source_columns.standardized_name as standardized_name",
            "source_columns.data_type as data_type",
            "source_columns.sql_transformation as sql_transformation"
        )
        return source_columns_schema_df

    def read_new_columns_schema(self):
        exploded_df = self.config_df.select(explode(self.config_df["schema"].new_columns).alias("new_columns"))
        new_columns_schema_df = exploded_df.selectExpr(
            "new_columns.name as name",
            "new_columns.data_type as data_type",
            "new_columns.sql_transformation as sql_transformation"
        )
        return new_columns_schema_df
    
    def read_column_descriptions_metadata(self):
        metadata_df = self.config_df.select("metadata.column_descriptions").alias("column_descriptions")
        descriptions_row_obj = metadata_df.first()["column_descriptions"]
        return descriptions_row_obj.asDict()
    
    def read_column_sequence_order(self):
        return list(self.config_df.first()["column_sequence_order"])

 

DataStandardizer 클래스 구현

class DataStandardizer:

    def __init__(self, raw_dp_path, temp_std_dp_path, std_dp_path):
        self.raw_dp_path = raw_dp_path
        self.temp_std_dp_path = temp_std_dp_path
        self.std_dp_path = std_dp_path

    def create_temp_std_dp_with_source_columns(self, source_columns_schema_df):
        source_columns_schema_df.createOrReplaceTempView("source_columns_config_table")
        select_query_sql = f"""
            SELECT 
                concat(
                    "SELECT ", 
                    array_join(collect_list(select_expression), ", "), 
                    " FROM delta.`{self.raw_dp_path}`"
                ) as select_query 
            FROM (
                SELECT 
                    CASE
                        WHEN sql_transformation = "" THEN concat("CAST(", concat("`", raw_name, "`"), " AS ", data_type, ") AS ", standardized_name)
                        ELSE concat("CAST(", sql_transformation, " AS ", data_type, ") AS ", standardized_name)
                    END as select_expression 
                FROM source_columns_config_table
            )
        """
        df = spark.sql(select_query_sql)
        select_query = df.first()["select_query"]
        create_sql_query = f"CREATE OR REPLACE TABLE delta.`{self.temp_std_dp_path}` as ( " + select_query + ")"
        spark.sql(create_sql_query)

    def add_new_columns_in_temp_std_dp(self, new_columns_schema_df):
        new_columns_schema_df_rows = new_columns_schema_df.collect()        
        for row in new_columns_schema_df_rows:
            add_new_columns_sql = f"ALTER TABLE delta.`{self.temp_std_dp_path}` ADD COLUMN {row['name']} {row['data_type']}"   
            sql_transformation = row["sql_transformation"].replace("{temp_std_dp_path}", self.temp_std_dp_path)  
            spark.sql(add_new_columns_sql)  
            spark.sql(sql_transformation)      
    
    def update_column_descriptions_metadata(self, column_descriptions_dict):
        for column_name, description in column_descriptions_dict.items():
            column_description_update_sql = f"ALTER TABLE delta.`{self.temp_std_dp_path}` CHANGE COLUMN {column_name} COMMENT '{description}';"
            spark.sql(column_description_update_sql)
        
    def move_data_to_std_dp(self, column_sequence_order):
        temp_std_df = spark.read.format("delta").load(self.temp_std_dp_path)
        temp_std_df = temp_std_df.select(column_sequence_order)
        temp_std_df.write.option("mergeSchema", "true").format("delta").mode("overwrite").save(self.std_dp_path)

    def run(self, config_reader):
        print("Raw df : ")
        raw_df = spark.read.format("delta").load(self.raw_dp_path)
        display(raw_df)

        source_columns_schema_df = config_reader.read_source_columns_schema()
        self.create_temp_std_dp_with_source_columns(source_columns_schema_df)

        new_columns_schema_df = config_reader.read_new_columns_schema()
        self.add_new_columns_in_temp_std_dp(new_columns_schema_df)

        column_descriptions_dict = config_reader.read_column_descriptions_metadata()
        self.update_column_descriptions_metadata(column_descriptions_dict)

        column_sequence_order = config_reader.read_column_sequence_order()
        self.move_data_to_std_dp(column_sequence_order)

        print("Standardized df : ")
        std_df = spark.read.format("delta").load(self.std_dp_path)
        display(std_df)

        print("Schema information for Standardized df : ")
        std_df.printSchema()    
        display(spark.sql(f"DESCRIBE TABLE delta.`{self.std_dp_path}`"))

`DataStandardizer` 클래스는 세 가지 속성을 사용합니다:
- `raw_dp_path`: 원본 데이터 제품 경로
- `temp_std_dp_path`: 표준화 작업을 위한 임시 경로
- `std_dp_path`: 최종 표준화된 데이터 제품 경로

각 메서드에 대한 설명은 다음과 같습니다:
1. create_temp_std_dp_with_source_columns — 원본 데이터 제품에서 직접 가져온 컬럼으로 표준화된 데이터 제품의 초기 버전을 생성합니다.
2. add_new_columns_in_temp_std_dp — 다른 데이터 제품과 조인하여 얻은 새로운 컬럼을 임시 표준화된 데이터 제품에 추가합니다.
3. update_column_descriptions_metadata — 각 컬럼에 대한 설명을 업데이트합니다.
4. move_data_to_std_dp— 임시/스테이징 영역에서 최종 표준화된 데이터 제품 경로로 데이터를 복사합니다.
5. run — 이 메서드는 위의 모든 단계를 조정하며, `ConfigReaderContract` 클래스를 사용하여 구현된 `config_reader` 인스턴스를 받아들입니다.

표준화된 데이터 제품 실행

다음은 `supplier` 원본 데이터 제품에서 프레임워크를 실행하는 코드입니다.

# 경로 정의
raw_dp_path = "dbfs:/FileStore/project/supplier"
std_dp_path = "dbfs:/FileStore/project/Product_Supplier"
temp_std_dp_path = "dbfs:/FileStore/project/Product_Supplier_temp"
config_path = "dbfs:/FileStore/project/supplier_config.json"

# 설정 리더 및 데이터 표준화 초기화
config_reader = ConfigReader(config_path)
data_standardizer = DataStandardizer(
    raw_dp_path=raw_dp_path,
    temp_std_dp_path=temp_std_dp_path,
    std_dp_path=std_dp_path
)

# DataStandardizer 클래스 실행
data_standardizer.run(config_reader)

 

위 이미지에서 볼 수 있듯이, 표준화된 DP는 설정 파일에 언급된 모든 열과 각 열에 대한 설명과 함께 생성됩니다.

예를 들어, 원본 DP의 sup_id 열은 Supplier_ID로 이름이 변경되었으며, 각 값 앞에 필요한 접두사(SUP)가 추가되었습니다. 설명 메타데이터인 "제품 공급업체의 고유 식별자"도 업데이트되었습니다. 또한, 원본 DP의 가격과 수량을 곱하여 Total_Cost 열이 추가되었습니다. 새로운 열인 Product_ID는 다른 데이터 제품인 Product와 조인을 통해 추가되었습니다.

이 블로그에서 사용된 코드는 이 GitHub 저장소에서 확인할 수 있습니다.

결론

Spark를 활용한 설정 기반 데이터 표준화 프레임워크는 원본 데이터를 고품질의 표준화된 데이터 제품으로 변환하는 데 매우 유용한 방법입니다. 설정 파일을 통해 표준화 규칙을 관리함으로써 유연성, 일관성, 유지 관리 용이성을 제공하며, 코드를 변경하지 않고도 동적으로 규칙을 조정할 수 있습니다. 또한, 데이터 품질 검증, 데이터 제품 버전 관리 및 기타 메타데이터 향상 기능을 쉽게 확장할 수 있어 매우 확장 가능하고 적용 범위가 넓습니다.

---

reference 

Config-Driven Data Standardization Framework using Spark