譯者 | 李睿

審校 | 重樓

本文介紹使用 Spark 進行數(shù)據(jù)處理的元數(shù)據(jù)和配置驅(qū)動的 Python 框架。這個強大的框架提供了一種精簡而靈活的方法來獲取文件、應用轉(zhuǎn)換和將數(shù)據(jù)加載到數(shù)據(jù)庫中。通過利用元數(shù)據(jù)和配置文件，該框架實現(xiàn)了高效且可擴展的數(shù)據(jù)處理管道。憑借其模塊化結(jié)構(gòu)，用戶可以輕松地根據(jù)其特定需求調(diào)整框架，確保與不同的數(shù)據(jù)源、文件格式和數(shù)據(jù)庫無縫集成。通過自動化流程和抽象復雜性，這一框架提高了生產(chǎn)力，減少了人工工作，并為數(shù)據(jù)處理任務提供了可靠的基礎。無論用戶是在進行大規(guī)模的數(shù)據(jù)處理還是頻繁的數(shù)據(jù)更新，該框架都使其能夠有效地利用Spark的力量，實現(xiàn)高效的數(shù)據(jù)集成、轉(zhuǎn)換和加載。

以下是一個元數(shù)據(jù)和配置驅(qū)動的Python框架的示例，該框架使用Spark進行數(shù)據(jù)處理，以攝取文件、轉(zhuǎn)換數(shù)據(jù)并將其加載到數(shù)據(jù)庫中。所提供的代碼是一個簡化的實現(xiàn)，用來說明這個概念。用戶可能需要調(diào)整它以適應其特定需求。

1.配置管理

配置管理部分處理加載和管理數(shù)據(jù)處理管道所需的配置設置。

• config.yaml：這個yaml文件包含配置參數(shù)和設置。以下是config.yaml文件的示例結(jié)構(gòu)：

YAML 
 input_paths:
 - /path/to/input/file1.csv
 - /path/to/input/file2.parquet
 database:
 host: localhost
 port: 5432
 user: my_user
 password: my_password
 database: my_database
 table: my_table

config.yaml文件包括以下元素：

• input_paths(列表)：指定輸入文件要處理的路徑?？梢栽诹斜碇邪鄠€文件路徑。
• 數(shù)據(jù)庫(字典)：包含數(shù)據(jù)庫連接信息。

o host：數(shù)據(jù)庫服務器的主機名或IP地址

o Port：連接數(shù)據(jù)庫的端口號

o user：身份驗證的用戶名

o Password：身份驗證的密碼

o database：數(shù)據(jù)庫名稱

o table：將加載轉(zhuǎn)換之后的數(shù)據(jù)的表名

用戶可以使用其他設置擴展此配置文件，例如Spark配置參數(shù)、日志記錄選項或特定于用戶的項目的任何其他配置。

• config.py：該模塊負責加載config.yaml文件

Python 
  # config.py
  import yaml

  def load_config():
 with open('config.yaml', 'r') as file:
 config = yaml.safe_load(file)
 return config

2.元數(shù)據(jù)管理

元數(shù)據(jù)管理部分處理輸入文件的元數(shù)據(jù)信息。它包括定義元數(shù)據(jù)結(jié)構(gòu)和管理元數(shù)據(jù)存儲庫。

• metadata.json：這個json文件包含每個輸入文件的元數(shù)據(jù)信息。以下是metadata.json文件的結(jié)構(gòu)示例：

YAML 
 {
 "/path/to/input/file1.csv": {
 "file_format": "csv",
 "filter_condition": "columnA > 10",
 "additional_transformations": [
 "transform1",
 "transform2"
 ]
 }
 "/path/to/input/file2.parquet": {
 "file_format": "parquet",
 "additional_transformations": [
 "transform3"
 ]
 }
 }

metadata.json文件包含以下元素：

每個輸入文件路徑是JSON對象中的鍵，對應的值是表示該文件元數(shù)據(jù)的字典。

• file_format：指定文件格式(例如csv、parquet等)。
• filter_condition(可選)：表示將應用于數(shù)據(jù)的過濾條件。在本例中，僅包括columnA大于10的行。
• additional_transforms(可選)：列出要應用于數(shù)據(jù)的其他轉(zhuǎn)換。用戶可以定義自己的轉(zhuǎn)換邏輯，并通過名稱引用它們。

用戶可以擴展元數(shù)據(jù)結(jié)構(gòu)，以包含其他相關信息，例如列名、數(shù)據(jù)類型、模式驗證規(guī)則等，具體取決于用戶的具體需求。

• metadata.py：這個模塊負責加載metadata.json文件

Python 
1 # metadata.py
2 import json
3
4 def load_metadata():
5 with open('metadata.json', 'r') as file:
6 metadata = json.load(file)
7 return metadata
8
9 def save_metadata(metadata):
10 with open('metadata.json', 'w') as file:
11 json.dump(metadata, file)
12

3.文件攝入

文件攝取部分負責將輸入文件攝取到Spark中進行處理。

• inclusion.py模塊掃描config.yaml文件中指定的輸入目錄，并檢索要處理的文件列表。
• 它檢查元數(shù)據(jù)存儲庫，以確定文件是否已經(jīng)被處理或是否需要任何更新。
• 使用Spark內(nèi)置的文件讀取器（例如Spark.read.csv、Spark.read.parquet等），它將文件加載到Spark DataFrames中。

Python 
 # ingestion.py
 from pyspark.sql import SparkSession

 def ingest_files(config):
 spark = SparkSession.builder.config("spark.sql.shuffle.partitions", "4").getOrCreate()

 for file_path in config['input_paths']:
 # Check if the file is already processed based on metadata
 if is_file_processed(file_path):
 continue

 # Read the file into a DataFrame based on metadata
 file_format = get_file_format(file_path)
 df = spark.read.format(file_format).load(file_path)

 # Perform transformations based on metadata
 df_transformed = apply_transformations(df, file_path)

 # Load transformed data into the database
 load_to_database(df_transformed, config['database'])

 # Update metadata to reflect the processing status
 mark_file_as_processed(file_path)

4.數(shù)據(jù)轉(zhuǎn)換

數(shù)據(jù)轉(zhuǎn)換部分處理基于元數(shù)據(jù)信息對輸入數(shù)據(jù)應用轉(zhuǎn)換。

• transformation.py模塊包含將轉(zhuǎn)換應用到Spark DataFrames的函數(shù)和邏輯。
• 從元數(shù)據(jù)存儲庫中讀取每個文件的元數(shù)據(jù)。
• 根據(jù)元數(shù)據(jù)，將所需的轉(zhuǎn)換應用到相應的Spark DataFrames。這可以包括過濾、聚合、連接等任務。
• 可以定義可重用的轉(zhuǎn)換函數(shù)或類來處理不同的文件格式或自定義轉(zhuǎn)換。
• 轉(zhuǎn)換后的Spark DataFrame返回進一步處理。

Python 
1 # transformations.py
2 def apply_transformations(df, file_path):
3 metadata = load_metadata()
4 file_metadata = metadata[file_path]
5 
6 # Apply transformations based on metadata
7 # Example: Filtering based on a condition
8 if 'filter_condition' in file_metadata:
9 df = df.filter(file_metadata['filter_condition'])
10 
11 # Add more transformations as needed
12 
13 return df
14

5.數(shù)據(jù)加載

數(shù)據(jù)加載部分側(cè)重于將轉(zhuǎn)換后的數(shù)據(jù)加載到指定的數(shù)據(jù)庫中。

• loading.py模塊包含用于建立與目標數(shù)據(jù)庫的連接并加載轉(zhuǎn)換后的數(shù)據(jù)的函數(shù)。
• 它從config.yaml文件中檢索數(shù)據(jù)庫連接的詳細信息。使用適當?shù)臄?shù)據(jù)庫連接器庫(例如，psycopg2,pyodbc等)，它建立到數(shù)據(jù)庫的連接。
• 轉(zhuǎn)換后的Spark DataFrame使用Spark的數(shù)據(jù)庫連接器（例如Spark.write.jdbc）寫入指定的數(shù)據(jù)庫表。
• 在加載完成后，關閉與數(shù)據(jù)庫的連接。

Python 
 # loading.py
 import psycopg2

 def load_to_database(df, db_config):
 conn = psycopg2.connect(
 host=db_config['host'],
 port=db_config['port'],
 user=db_config['user'],
 password=db_config['password'],
 database=db_config['database']
 )

 # Write DataFrame to a database table
 df.write \
 .format('jdbc') \
 .option('url', f"jdbc:postgresql://{db_config['host']}:{db_config['port']}/{db_config['database']}") \
 .option('dbtable', db_config['table']) \
 .option('user', db_config['user']) \
 .option('password', db_config['password']) \
 .mode('append') \
 .save()
 
 conn.close()

6.執(zhí)行流程

執(zhí)行流部分編排整個數(shù)據(jù)處理管道。

• main.py模塊作為框架的入口點。
• 它從config.yaml文件加載配置設置。
• 它從元數(shù)據(jù)存儲庫中檢索元數(shù)據(jù)。
• 使用Spark調(diào)用文件攝取模塊來處理輸入文件。
• 使用數(shù)據(jù)加載模塊將轉(zhuǎn)換后的數(shù)據(jù)加載到數(shù)據(jù)庫中。
• 更新元數(shù)據(jù)存儲庫，以反映每個文件的處理狀態(tài)。
• 根據(jù)需要實現(xiàn)額外的錯誤處理、日志記錄和監(jiān)視。

Python 
 # main.py
 import config
 import metadata
 import ingestion

 # Load configuration and metadata
 config_data = config.load_config()
 metadata_data = metadata.load_metadata()

 # Process files using Spark
 ingestion.ingest_files(config_data)

 # Save updated metadata
 metadata.save_metadata(metadata_data)

7. CLI或UI界面(可選)

CLI或UI界面部分提供了一種用戶友好的方式與框架進行交互。

• cli.py模塊使用類似argparse的庫創(chuàng)建了一個命令行界面（cli）。
• 用戶可以通過提供配置文件的路徑作為參數(shù)，從命令行運行框架。
• CLI解析提供的參數(shù)，加載配置和元數(shù)據(jù)，并觸發(fā)數(shù)據(jù)處理管道。
• 可以根據(jù)需要向接口添加其他功能，例如查看日志、指定輸入/輸出路徑或監(jiān)視管道。

Python 
 # cli.py
 import argparse
 import config
 import metadata
 import ingestion

 parser = argparse.ArgumentParser(description='Data Processing Framework')

 def main():
 parser.add_argument('config_file', help='Path to the configuration file')
 args = parser.parse_args()
 
 # Load configuration and metadata
 config_data = config.load_config(args.config_file)
 metadata_data = metadata.load_metadata()
 
 # Process files using Spark
 ingestion.ingest_files(config_data)
 
  # Save updated metadata
 metadata.save_metadata(metadata_data)
 
 if __name__ == '__main__':
 main()

使用更新的main（）函數(shù)，用戶可以通過提供配置文件的路徑作為參數(shù)，從命令行運行框架。例如：

Shell 
 python cli.py my_config.yaml

這將基于所提供的配置文件執(zhí)行數(shù)據(jù)處理管道。

注意：此代碼是一個簡化的示例，用戶需要根據(jù)自己的特定需求對其進行定制。此外，可能需要處理錯誤情況，添加日志記錄，并修改代碼以適合其特定數(shù)據(jù)庫連接器庫（例如，psycopg2、pyodbc等）。

需要注意的是，所提供的說明概述了框架的結(jié)構(gòu)和主要組成部分。用戶需要根據(jù)其需求以及選擇使用的庫和工具，在每個模塊中實現(xiàn)特定的邏輯和細節(jié)。

總之，元數(shù)據(jù)和配置驅(qū)動的Python數(shù)據(jù)處理框架與Spark提供了一個全面的解決方案來處理復雜的數(shù)據(jù)處理任務。通過利用元數(shù)據(jù)和配置文件，該框架提供了靈活性和可擴展性，使用戶能夠無縫集成各種數(shù)據(jù)源、應用轉(zhuǎn)換并將數(shù)據(jù)加載到數(shù)據(jù)庫中。憑借其模塊化設計，用戶可以輕松定制和擴展框架，以滿足其特定需求。通過自動化數(shù)據(jù)處理流程，這個框架使用戶能夠提高生產(chǎn)力，減少人工工作，并確保數(shù)據(jù)處理工作流程的一致性和可靠性。無論用戶是處理大量數(shù)據(jù)還是頻繁更新數(shù)據(jù)集，該框架都使用戶能夠使用Spark的強大功能高效地處理、轉(zhuǎn)換和加載數(shù)據(jù)，并實現(xiàn)更好的洞察力和決策能力。

原文標題：Metadata and Config-Driven Python Framework for Big Data Processing Using Spark，作者：Amlan Patnaik