So erstellen Sie ETL-Pipelines in Python

Veröffentlicht: 2022-01-11

ETL steht für Extrahieren , Transformieren , Laden. Als Teil des ETL-Prozesses werden Daten extrahiert, transformiert und in Data Warehouses geladen, damit Unternehmen sie analysieren können, um strategische Entscheidungen zu treffen.

Im Folgenden sind die wichtigsten Schritte aufgeführt, die in der ETL-Pipeline durchgeführt werden:

Extrahieren: Dieser Prozess sammelt und integriert Daten aus einer Vielzahl von Quellen, einschließlich Datenbanken, Data Lakes, CRMs und anderen.
Transformation: Dies ist die wichtigste Phase in einer ETL-Pipeline. Um Daten analysebereit zu machen, müssen sie in diesem Schritt richtig gesammelt, sortiert, bereinigt und geschwenkt werden.
Laden: Bei diesem Prozess werden strukturierte oder unstrukturierte Daten aus Data Lakes, Datenbanken und anderen Quellen in Data Warehouses importiert, damit Datenanalysten oder andere Benutzer auf einfache Weise tiefe Einblicke gewinnen können.

Die Bedeutung von Python ETL verstehen

Python ist eine der beliebtesten und am häufigsten genutzten Programmiersprachen der modernen Welt mit endlosen Anwendungen in einer Vielzahl von Bereichen. Es hat die renommierte Auszeichnung TIOBE Programming Language of the Year 2021 gewonnen.

Die flexible und dynamische Natur von Python macht es ideal für Bereitstellungs-, Analyse- und Wartungsaufgaben. Python ETL ist eine der entscheidenden Fähigkeiten, die im Data Engineering erforderlich sind, um Datenpipelines zu erstellen, statistische Modelle zu entwickeln und diese gründlich zu analysieren.

Es hat sich aufgrund seiner Benutzerfreundlichkeit und robusten Bibliotheken für den Zugriff auf Datenbanken und Speichersysteme zu einem beliebten Tool zum Ausführen von ETL-Prozessen entwickelt. Viele Teams verwenden Python für ETL & Data Engineering anstelle eines ETL-Tools, da es für diese Aufgaben vielseitiger und leistungsfähiger ist.

Der größte Vorteil von Python gegenüber anderen Programmiersprachen ist die einfache Verwendung in Data Mining, Data Science, Big Data, künstlicher Intelligenz und maschinellem Lernen.

Unternehmen auf der ganzen Welt nutzen Python für ihre Daten, um Erkenntnisse zu gewinnen, ihre Abläufe zu verwalten und dafür zu sorgen, dass alles reibungslos läuft.

2 einfache Schritte zum Erstellen einer Python-ETL-Pipeline

In diesem Teil lernen Sie die wesentlichen Schritte zum Erstellen einer ETL-Pipeline mit Python kennen . Sie erstellen eine einfache Datenpipeline, die Daten aus MySQL- und Microsoft SQL Server-Datenbanken in eine Microsoft SQL Server-Datenbank einspeist.

Führen Sie die folgenden Schritte aus, um das Python-ETL-Skript einzurichten:

Schritt 1: Installieren Sie die erforderlichen Module

Um die Python-ETL-Pipeline einzurichten, müssen Sie die folgenden Module installieren:

Python-zu-MySQL-Connector: mysql-connector-python (Verwenden Sie zum Installieren den Befehl pip install mysql-connector-python )
Python zu Microsoft SQL Server Connector: pyodbc (Verwenden Sie zum Installieren den Befehl pip install pyodbc )

Schritt 2: Richten Sie das ETL-Verzeichnis ein

Nach der Installation der oben genannten Pakete müssen Sie 4 Python-Dateien erstellen, die unten in Ihrem Projektverzeichnis erwähnt werden:

db_credentials.py: Diese Datei enthält Code zum Herstellen von Verbindungen mit allen Datenbanken.
sql_queries.py: Diese Datei enthält die häufig verwendeten Datenbankabfragen zum Extrahieren und Laden von Daten im Zeichenfolgenformat.
etl.py: Diese Datei enthält die notwendigen Operationen, um eine Verbindung zur Datenbank herzustellen und die erforderlichen Abfragen auszuführen.
main.py: Dies ist die primäre Datei, die den Ablauf und die Ausführung der Python-ETL-Pipeline regelt.

A) db_credentials.py

Alle Quell- und Zieldatenbank-Verbindungszeichenfolgen sollten in dieser Datei enthalten sein. Es sollte alle notwendigen Informationen für den Zugriff auf die relevante Datenbank in einem Listenformat enthalten, damit es bei Bedarf schnell iteriert werden kann. Im Folgenden finden Sie ein Beispiel für ein Python-Skript zum Herstellen der Datenbankverbindung:

 datawarehouse_name = 'your_dwh_name'
# SQL-Server (Zieldatenbank, Datawarehouse)
datawarehouse_db_config = {
  'Trusted_Connection': 'ja',
  'Treiber': '{SQL Server}',
  'server': 'datawarehouse_sql_server',
  'Datenbank': '{}'.format(datawarehouse_name),
  'user': 'your_db_uname',
  'password': 'your_db_pword',
  'autocommit': Richtig,
}
# Quelldatenbank > SQL-Server
sqlserver_db_config = [
  {
    'Trusted_Connection': 'ja',
    'Treiber': '{SQL Server}',
    'server': 'your_db_sql_server',
    'Datenbank': 'db_1st',
    'user': 'your_db_uname',
    'password': 'your_db_pword',
    'autocommit': Richtig,
  }
]
# Quelldatenbank > mysql
mysql_db_config = [
  {
    'Benutzer': 'Ihr_1_Benutzer',
    'password': 'your_1_pword',
    'host': 'db_connection_string_1',
    'Datenbank': 'db_1st',
  },
  {
    'Benutzer': 'Ihr_2_Benutzer,
    'Passwort': 'Ihr_2_Passwort',
    'host': 'db_connection_string_2',
    'Datenbank': 'db_2nd',
  },
]

B) sql_queries.py

Diese Datei enthält Abfragen zum Extrahieren von Daten aus den Quelldatenbanken und zum Laden in die Zieldatenbank. Das folgende Skript hilft Ihnen bei dieser Aufgabe:

 # Beispielabfragen, werden für verschiedene Datenbankplattformen eindeutig sein

sqlserver_extract = ('''
  SELECT sqlserver_col_1, sqlserver_col_2, sqlserver_col_3
  VON sqlserver_1_table
''')
sqlserver_insert = ('''
  INSERT INTO table_demo (Spalte_1, Spalte_2, Spalte_3)
  WERTE (?, ?, ?)  
''')
mysql_extract = ('''
  WÄHLEN Sie mysql_col_1, mysql_col_2, mysql_col_3
  AUS mysql_demo_table
''')
mysql_insert = ('''
  INSERT INTO table_demo (Spalte_1, Spalte_2, Spalte_3)
  WERTE (?, ?, ?)  
''')

# Abfragen werden exportiert
Klasse Sql_Query:
  def __init__(selbst, extrahieren_abfrage, laden_abfrage):
    self.extract_query = extrahieren_abfrage
    self.load_query = Lade_Abfrage   
# Instanzen für die Sql_Query-Klasse erstellen
sqlserver_query = SqlQuery(sqlserver_extract, sqlserver_insert)
mysql_query = SqlQuery(mysql_extract, mysql_insert)
# Erstellen einer Liste zum Durchlaufen von Werten
mysql_queries = [mysql_abfrage]
sqlserver_queries = [sqlserver_abfrage]

C) etl.py

Diese Datei sollte den Code enthalten, der für den Zugriff auf die relevanten Datenbanken und die Ausführung der erforderlichen Abfragen erforderlich ist. Das folgende Skript hilft Ihnen bei dieser Aufgabe:

 # Python-basierte Module
pyodbc importieren
mysql.connector importieren

def etl(Abfrage, Quelle_cnx, Ziel_cnx):
  # Daten aus der Demo-Quelldatenbank extrahieren
  source_cursor = source_cnx.cursor()
  source_cursor.execute(query.extract_query)
  data = source_cursor.fetchall()
  source_cursor.close()

  # Daten in die Demo Data Warehouse-Datenbank laden
  
wenn Daten:
    target_cursor = target_cnx.cursor()
    target_cursor.execute("USE {}".format(name_for_datawarehouse))
    target_cursor.executemany(query.load_query, Daten)
    print('Daten in die Demo-Datenbank des Data Warehouse geladen')
    target_cursor.close()
  anders:
    print('Daten sind leer')

def etl_process (Abfragen, target_cnx, source_db_config, db_platform):

  # Konfigurieren der Verbindung zur Demo-Quelldatenbank
  if db_platform == 'mysql':
    source_cnx = mysql.connector.connect(**source_db_config)
  elif db_platform == 'sqlserver':
    source_cnx = pyodbc.connect(**source_db_config)
  anders:
    Rückgabe 'Fehler! unbekannte Quelldatenbankplattform'
  # SQL-Abfragen durchlaufen
  für Abfrage in Abfragen:
    etl (Abfrage, source_cnx, target_cnx)    
  # Schließen Sie die Quelldatenbankverbindung
  source_cnx.close()

D) main.py

Diese Datei enthält Code zum Durchlaufen der angegebenen Anmeldeinformationen, um eine Verbindung zur Datenbank herzustellen und die erforderlichen ETL-Python-Vorgänge auszuführen. Das folgende Skript hilft Ihnen bei dieser Aufgabe:

 # Variablen
aus db_credentials import datawarehouse_db_config, sqlserver_db_config, mysql_db_config
aus sql_queries import sqlserver_queries, mysql_queries

# Methoden
aus etl import etl_process
def main():
  print('starting the etl data process')
	
  # Verbindung für SQL Server herstellen, gewünschter Zielspeicher
  target_cnx = pyodbc.connect(**datawarehouse_db_config)
	
  # Anmeldeinformationen durchlaufen
  # Datenbank > mysql
  für config in mysql_db_config: 
    Versuchen:
      print("lade db: " + config['database'])
      etl_process(mysql_queries, target_cnx, config, 'mysql')
    außer Ausnahme als Fehler:
      print("etl for {} has error".format(config['database']))
      print('Fehlermeldung: {}'.format(error))
      fortsetzen
	
  # Datenbank > SQL-Server
  für config in sqlserver_db_config: 
    Versuchen:
      print("lade db: " + config['database'])
      etl_process(sqlserver_queries, target_cnx, config, 'sqlserver')
    außer Ausnahme als Fehler:
      print("etl for {} has error".format(config['database']))
      print('Fehlermeldung: {}'.format(error))
      fortsetzen

  target_cnx.close()
if __name__ == "__main__":
  hauptsächlich()

Fazit

Gute Arbeit! Sie haben sich erfolgreich ein grundlegendes Verständnis für den Aufbau einer Python ETL-Pipeline angeeignet. Jetzt können Sie Ihr benutzerdefiniertes Python-ETL-Skript basierend auf Ihren Anforderungen implementieren, indem Sie Änderungen an den verwendeten Datenbanken und entsprechenden Abfragen vornehmen.

Um die weit verbreiteten Python-ETL-Tools in der Branche zu erkunden, lesen Sie den Best Python ETL Tools-Blog.

Die meisten Organisationen arbeiten heutzutage mit Big Data. Daher kann die Erstellung einer ETL-Pipeline von Grund auf für solche Daten zeitaufwändig und herausfordernd sein.

Darüber hinaus müssen Unternehmen eine erhebliche Menge an Ressourcen investieren, um es zu erstellen und dann sicherzustellen, dass sie mit dem hohen Datenvolumen und den Schemaschwankungen Schritt halten können.

Anstatt also ETL-Skripte von Grund auf neu zu erstellen, können Sie automatisierte Datenpipelines wie Hevo nutzen.

Haben Sie irgendwelche Gedanken dazu? Lassen Sie es uns unten in den Kommentaren wissen oder übertragen Sie die Diskussion auf unseren Twitter oder Facebook.