كيفية بناء خطوط أنابيب ETL في بايثون

نشرت: 2022-01-11

يرمز ETL إلى E xtract و T ransform و L oad. كجزء من عملية ETL ، يتم استخراج البيانات وتحويلها وتحميلها في مستودعات البيانات حتى تتمكن المؤسسات من تحليلها لاتخاذ قرارات إستراتيجية.

فيما يلي الخطوات الرئيسية التي تم إجراؤها في خط أنابيب ETL:

استخراج: تقوم هذه العملية بجمع وتكامل البيانات من مجموعة متنوعة من المصادر ، بما في ذلك قواعد البيانات وبحيرات البيانات و CRMs وغيرها.
التحويل: هذه هي المرحلة الأكثر أهمية في خط أنابيب ETL. لجعل البيانات جاهزة للتحليلات ، يجب جمعها وفرزها وتنظيفها وتمحورها بشكل صحيح في هذه الخطوة.
التحميل: تتضمن هذه العملية استيراد بيانات منظمة أو غير منظمة من بحيرات البيانات وقواعد البيانات ومصادر أخرى إلى مستودعات البيانات بحيث يمكن لمحللي البيانات أو المستخدمين الآخرين اكتساب رؤى عميقة بسهولة.

فهم أهمية Python ETL

Python هي واحدة من لغات البرمجة الأكثر شهرة وشائعة في العالم ، مع تطبيقات لا حصر لها في مجموعة متنوعة من المجالات. وقد فازت بجائزة TIOBE لأفضل لغة برمجة لعام 2021.

إن طبيعة Python المرنة والديناميكية تجعلها مثالية لمهام النشر والتحليل والصيانة. تعد Python ETL واحدة من المهارات الأساسية المطلوبة في هندسة البيانات لبناء خطوط أنابيب البيانات ، وتطوير النماذج الإحصائية ، وإجراء تحليل شامل لها.

لقد أصبحت أداة شائعة لتنفيذ عمليات ETL نظرًا لسهولة استخدامها ومكتباتها القوية للوصول إلى قواعد البيانات وأنظمة التخزين. تستخدم العديد من الفرق Python لـ ETL وهندسة البيانات بدلاً من أداة ETL لأنها أكثر تنوعًا وقوة لهذه المهام.

أكبر فائدة لبايثون على لغات البرمجة الأخرى هي بساطة الاستخدام في التنقيب عن البيانات وعلوم البيانات والبيانات الضخمة والذكاء الاصطناعي والتعلم الآلي.

تستخدم الشركات في جميع أنحاء العالم Python لبياناتها للحصول على رؤى وإدارة عملياتها والحفاظ على سير كل شيء بسلاسة.

خطوتان سهلتان لبناء خط أنابيب Python ETL

في هذا الجزء ، ستتعلم الخطوات الأساسية لبناء خط أنابيب ETL باستخدام Python . ستنشئ خط أنابيب بيانات أساسيًا يغذي البيانات في قاعدة بيانات Microsoft SQL Server من قواعد بيانات MySQL و Microsoft SQL Server.

لإعداد برنامج Python ETL النصي ، اتبع الخطوات أدناه:

الخطوة 1: قم بتثبيت الوحدات المطلوبة

لإعداد Python ETL Pipeline ، ستحتاج إلى تثبيت الوحدات النمطية التالية:

Python to MySQL Connector: mysql-connector-python (استخدم الأمر pip install mysql-connector-python للتثبيت)
Python إلى Microsoft SQL Server Connector: pyodbc (استخدم الأمر pip install pyodbc للتثبيت)

الخطوة 2: قم بإعداد دليل ETL

بعد تثبيت الحزم أعلاه ، تحتاج إلى إنشاء 4 ملفات Python ، مذكورة أدناه في دليل المشروع الخاص بك:

db_credentials.py: يتضمن هذا الملف رمزًا لإنشاء اتصالات مع كافة قواعد البيانات.
sql_queries.py: يشتمل هذا الملف على استعلامات قاعدة البيانات شائعة الاستخدام لاستخراج البيانات وتحميلها بتنسيق سلسلة.
etl.py: يمتلك هذا الملف العمليات اللازمة للاتصال بقاعدة البيانات وتشغيل الاستعلامات المطلوبة.
main.py: هذا هو الملف الأساسي الذي ينظم تدفق وتنفيذ Python ETL Pipeline.

أ) db_credentials.py

يجب تضمين كافة سلاسل اتصال قاعدة البيانات المصدر والهدف في هذا الملف. يجب أن يحتوي على جميع المعلومات اللازمة للوصول إلى قاعدة البيانات ذات الصلة في شكل قائمة بحيث يمكن تكرارها بسرعة عند الحاجة. فيما يلي نموذج لبرنامج نصي من Python لتأسيس اتصال قاعدة البيانات:

 datawarehouse_name = 'your_dwh_name'
# خادم SQL (الهدف ديسيبل ، مستودع البيانات)
مستودع البيانات datawarehouse_db_config = {
  "Trusted_Connection": "نعم" ،
  'سائق': '{SQL Server}'،
  "الخادم": "datawarehouse_sql_server" ،
  'قاعدة البيانات': تنسيق '{}' (datawarehouse_name) ،
  "المستخدم": "your_db_uname" ،
  "كلمة المرور": "your_db_pword" ،
  "الالتزام التلقائي": صحيح ،
}
# المصدر ديسيبل> خادم SQL
sqlserver_db_config = [
  {
    "Trusted_Connection": "نعم" ،
    'سائق': '{SQL Server}'،
    "الخادم": "your_db_sql_server" ،
    'قاعدة البيانات': 'db_1st'،
    "المستخدم": "your_db_uname" ،
    "كلمة المرور": "your_db_pword" ،
    "الالتزام التلقائي": صحيح ،
  }
]
# المصدر db> mysql
mysql_db_config = [
  {
    "المستخدم": "your_1_user"،
    "كلمة المرور": "your_1_pword"،
    "host": "db_connection_string_1"،
    'قاعدة البيانات': 'db_1st'،
  } ،
  {
    "المستخدم": "your_2_user،
    "كلمة المرور": "your_2_password"،
    "المضيف": "db_connection_string_2" ،
    'قاعدة البيانات': 'db_2nd'،
  } ،
]

ب) sql_queries.py

يتضمن هذا الملف استعلامات لاستخراج البيانات من قواعد البيانات المصدر وتحميلها في قاعدة البيانات الهدف. سيساعدك البرنامج النصي التالي على أداء هذه المهمة:

 # استعلامات المثال ، ستكون فريدة لأنظمة قواعد البيانات المختلفة

sqlserver_extract = (""
  حدد sqlserver_col_1 و sqlserver_col_2 و sqlserver_col_3
  من sqlserver_1_table
"")
sqlserver_insert = (""
  INSERT INTO table_demo (col_1 ، col_2 ، col_3)
  القيم (؟، ؟، ؟)  
"")
mysql_extract = (""
  حدد mysql_col_1 و mysql_col_2 و mysql_col_3
  من mysql_demo_table
"")
mysql_insert = (""
  INSERT INTO table_demo (col_1 ، col_2 ، col_3)
  القيم (؟، ؟، ؟)  
"")

# من الاستفسارات التي يتم تصديرها
فئة Sql_Query:
  def __init __ (self، extract_query، load_query):
    self.extract_query = extract_query
    self.load_query = load_query   
# إنشاء مثيلات لفئة Sql_Query
sqlserver_query = SqlQuery (sqlserver_extract ، sqlserver_insert)
mysql_query = SqlQuery (mysql_extract، mysql_insert)
# إنشاء قائمة للتكرار عبر القيم
mysql_queries = [mysql_query]
sqlserver_queries = [sqlserver_query]

ج) etl.py

يجب أن يتضمن هذا الملف الكود المطلوب للوصول إلى قواعد البيانات ذات الصلة وتنفيذ الاستعلامات المطلوبة. سيساعدك البرنامج النصي التالي على أداء هذه المهمة:

 # وحدات قائمة على بيثون
استيراد ملفات pyodbc
استيراد mysql.connector

def etl (استعلام ، source_cnx ، target_cnx):
  # استخراج البيانات من قاعدة بيانات المصدر التجريبي
  source_cursor = source_cnx.cursor ()
  source_cursor.execute (query.extract_query)
  البيانات = source_cursor.fetchall ()
  source_cursor.close ()

  # تحميل البيانات في مستودع البيانات التجريبي ديسيبل
  
إذا كانت البيانات:
    target_cursor = target_cnx.cursor ()
    target_cursor.execute ("USE {}". تنسيق (name_for_datawarehouse))
    target_cursor.executemany (query.load_query، data)
    طباعة ("البيانات التي تم تحميلها إلى مستودع البيانات التجريبي ديسيبل")
    target_cursor.close ()
  آخر:
    طباعة ("البيانات فارغة")

def etl_process (الاستعلامات ، target_cnx ، source_db_config ، db_platform):

  # تكوين اتصال قاعدة بيانات المصدر التجريبي
  إذا كانت db_platform == 'mysql':
    source_cnx = mysql.connector.connect (** source_db_config)
  elif db_platform == 'sqlserver':
    source_cnx = pyodbc.connect (** source_db_config)
  آخر:
    العودة خطأ! منصة قاعدة بيانات المصدر غير المعترف بها "
  # حلقة من خلال استعلامات SQL
  للاستعلام في الاستعلامات:
    etl (استعلام ، source_cnx ، target_cnx)    
  # إغلاق اتصال المصدر ديسيبل
  source_cnx.close ()

د) main.py

يتضمن هذا الملف رمزًا للتكرار من خلال بيانات اعتماد معينة للاتصال بقاعدة البيانات وتنفيذ عمليات ETL Python الضرورية. سيساعدك البرنامج النصي التالي على أداء هذه المهمة:

 # المتغيرات
من db_credentials استيراد datawarehouse_db_config، sqlserver_db_config، mysql_db_config
من sql_queries import sqlserver_queries، mysql_queries

# طرق
من etl استيراد etl_process
def main ():
  طباعة ("بدء معالجة بيانات etl")
	
  # إنشاء اتصال لـ SQL Server ، وجهة التخزين المرغوبة
  target_cnx = pyodbc.connect (** مستودع البيانات_db_config)
	
  # التكرار من خلال بيانات الاعتماد
  # قاعدة بيانات> mysql
  للتكوين في mysql_db_config: 
    يحاول:
      طباعة ("تحميل ديسيبل:" + التكوين ['قاعدة بيانات'])
      etl_process (mysql_queries، target_cnx، config، 'mysql')
    باستثناء الاستثناء كخطأ:
      طباعة ("يوجد خطأ في etl لـ {}" .format (config ['database']))
      طباعة ('رسالة الخطأ: {}'. تنسيق (خطأ))
      استمر
	
  # قاعدة بيانات> خادم SQL
  للتكوين في sqlserver_db_config: 
    يحاول:
      طباعة ("تحميل ديسيبل:" + التكوين ['قاعدة بيانات'])
      etl_process (sqlserver_queries، target_cnx، config، 'sqlserver')
    باستثناء الاستثناء كخطأ:
      طباعة ("يوجد خطأ في etl لـ {}" .format (config ['database']))
      طباعة ('رسالة الخطأ: {}'. تنسيق (خطأ))
      استمر

  target_cnx.close ()
إذا __name__ == "__main__":
  رئيسي()

خاتمة

عمل عظيم! لقد اكتسبت بنجاح فهمًا أساسيًا لبناء Python ETL Pipeline. يمكنك الآن تنفيذ برنامج Python ETL المخصص بناءً على متطلباتك عن طريق إجراء تغييرات على قواعد البيانات المستخدمة والاستعلامات وفقًا لذلك.

لاستكشاف أدوات Python ETL المستخدمة على نطاق واسع في الصناعة ، اقرأ مدونة Best Python ETL Tools.

تعمل معظم المؤسسات في الوقت الحاضر باستخدام البيانات الضخمة. ومن ثم ، فإن إنشاء خط أنابيب ETL من البداية لمثل هذه البيانات يمكن أن يكون مضيعة للوقت وصعبًا.

علاوة على ذلك ، ستحتاج المؤسسات إلى استثمار قدر كبير من الموارد من أجل بنائه ومن ثم ضمان قدرتها على مواكبة حجم البيانات الكبير وتقلبات المخطط.

لذلك ، بدلاً من إنشاء نصوص ETL من البداية ، يمكنك الاستفادة من خطوط أنابيب البيانات الآلية مثل Hevo.

هل لديك أي أفكار حول هذا؟ أخبرنا أدناه في التعليقات أو انقل المناقشة إلى Twitter أو Facebook.