Snowflake Connector for Python: In wenigen Minuten startklar

Der Snowflake Connector ist ein natives Python-Paket und lässt sich daher wie jedes andere Python-Paket installieren.

Mit pip

Am einfachsten installieren Sie die Bibliothek über pip, das offizielle Paketverwaltungssystem für Python:

1pip install snowflake-connector-python

Mit conda

Wenn Sie conda als Paketmanager verwenden, installieren Sie das Paket so:

1conda install snowflake-connector-python

Hier sehen Sie ein vollständiges End-to-End-Beispiel, wie Sie mit dem Snowflake Connector eine Query ausführen und die Ergebnisse weiterverarbeiten. Es zeigt den gesamten Credit-Verbrauch je Virtual Warehouse in den letzten 30 Tagen:

Die Ausgabe sieht in etwa so aus:

('REPLICATOR_LARGE', Decimal('669.080833333'))
('REPLICATOR_XSMALL', Decimal('613.355000002'))
('REPLICATOR_XLARGE', Decimal('416.437500000'))
('BACKEND_LARGE', Decimal('162.702500031'))
('BACKEND', Decimal('105.263888856'))
('BACKEND_MEDIUM', Decimal('47.033055570'))
('REPLICATOR_MEDIUM', Decimal('40.775833333'))
('REPLICATOR_SMALL', Decimal('31.046388890'))
('BACKEND_STAGING', Decimal('26.985555532'))
('BACKEND_SMALL', Decimal('12.798333345'))
('SELECT_DOGFOOD', Decimal('11.283055542'))
('ADMIN', Decimal('10.899444408'))
('BACKEND_STAGING_MEDIUM', Decimal('6.822222224'))
('BI_XSMALL', Decimal('1.814999976'))
('CLOUD_SERVICES_ONLY', Decimal('0E-9'))

Wie Sie Ihren Account Identifier ermitteln, lesen Sie in diesem Beitrag.

Connection- und Cursor-Objekte

Wie im Quickstart-Beispiel oben zu sehen, arbeiten wir mit zwei getrennten Objekten: dem Connection- und dem Cursor-Objekt.

Das Connection-Objekt kümmert sich um Authentifizierung und Verbindungsaufbau zu Snowflake und liefert Cursor zurück. Über die einzelnen Cursor führen Sie dann Queries gegen Snowflake aus und holen Ergebnisse ab. Eine einzelne Connection kann mehrere Cursor haben. Denselben Cursor können Sie für mehrere Queries wiederverwenden – Sie müssen nicht jedes Mal einen neuen anlegen. Gleiches gilt für das Connection-Objekt: Da der Verbindungsaufbau zu Snowflake einige Sekunden dauert, sollten Sie ihn nur einmal durchführen.

Mehr zu den verfügbaren Methoden beider Objekte finden Sie in der offiziellen API-Dokumentation des Snowflake Python Connectors.

Synchrone vs. asynchrone Queries

Mit dem Snowflake Connector for Python lassen sich Queries auf zwei Arten ausführen: synchron und asynchron.

Eine synchrone Query absetzen

Bei einer synchronen Query wartet Ihr Python-Prozess, bis das Ergebnis zurückgeliefert wird. Aufgrund der Einfachheit ist das der gängigste Ansatz beim Arbeiten mit dem Python Connector:

cur = conn.cursor()
cur.execute('select * from table') # Python wartet, bis die Query fertig ist, bevor es weitergeht
results = cur.fetchall()

Eine asynchrone Query absetzen

Bei einer asynchronen Query gibt das Snowflake-Python-Paket sofort die Kontrolle an Ihren Python-Prozess zurück, noch bevor die Query abgeschlossen ist. Das ist besonders praktisch, wenn Sie Multithread-Anwendungen in Python entwickeln (etwa eine Webanwendung) oder mit Python viele Queries parallel ausführen.

Für eine asynchrone Query nutzen Sie statt der üblichen execute-Methode die Cursor-Methode execute_async. An späterer Stelle im Code müssen Sie Snowflake per Polling abfragen, ob die Query bereits abgeschlossen ist.

cur = conn.cursor()
cur.execute_async('select * from table') # Python geht sofort zur nächsten Zeile über
query_id = cur.sfqid

while True:
		if conn.is_still_running(conn.get_query_status(query_id)):
				break
    time.sleep(1)

cur.get_results_from_sfqid(query_id)
results = cur.fetchall()

Transaktionssteuerung

Standardmäßig werden Queries, die Sie über den Snowflake Connector ausführen, automatisch committet.

Für eine feinere Kontrolle über Ihre SQL-Transaktionen können Sie beim Verbindungsaufbau autocommit=False setzen.

Mit dem with-Kontextmanager

Sehen Sie sich folgenden Code an: Schlägt eines der SQL-Statements fehl, rollt Snowflake sie automatisch zurück. Sind sie erfolgreich, werden sie automatisch committet, sobald der with-Block verlassen wird.

with snowflake.connector.connect(
    user=USER,
    password=PASSWORD,
    account=ACCOUNT,
    autocommit=False,
) as conn:
    cur = conn.cursor()
    cur.execute("insert into table values(...)")
    cur.execute("insert into table values(...)")

Manuelle Steuerung mit try-except

Für noch feinere Kontrolle können Sie SQL-Transaktionen manuell committen oder zurückrollen – mit den Methoden conn.commit() und conn.close() sowie dem try-except-finally-Pattern:

con = snowflake.connector.connect(
  user=USER,
  password=PASSWORD,
  account=ACCOUNT,
  autocommit=False)
try:
    cur = conn.cursor()
    cur.execute("insert into table values(...)")
    cur.execute("insert into table values(...)")
    conn.commit()
except Exception as e:
    conn.rollback()
    raise e
finally:
    conn.close()

Verbindung schließen

Als Best Practice sollten Sie Ihre Verbindung am Ende des Python-Skripts mit conn.close() schließen. Damit das selbst bei Fehlern zuverlässig passiert, packen Sie Ihren Code in einen try - finally-Block:

try:
   cur.execute('...')
   results = cur.fetchall()
   # ...
except:
    # Fehler nach Bedarf behandeln
finally:
		conn.close()

Ein weiteres nützliches Pattern ist das Python-with()-Konstrukt, das die Verbindung automatisch schließt:

with snowflake.connector.connect(...) as conn:
		cur = conn.cursor()
		cur.execute('...')
		results = cur.fetchall()

# Die Methode conn.close() wird nach dem with-Block automatisch aufgerufen
# Weitere Operationen ohne Snowflake-Bezug
with open('filename.csv', 'w') as f:
		f.write(...)

Schauen wir uns weitere Beispiele an, die typische Patterns mit dem Snowflake Python Connector zeigen.

Parameter an Ihr SQL-Statement übergeben

Häufig möchten Entwickler unterschiedliche Parameter an ihre SQL-Statements übergeben.

Python bringt mit f-Strings hervorragende Mittel zur String-Manipulation mit, die sich genau dafür anbieten:

user_id = 123456
sql = f"select * from users where user_id={user_id}"
cur.execute(sql)

Alternativ können Sie Bind-Variablen nutzen:

warehouse_name='ADMIN'
sql="""
select *
from snowflake.account_usage.warehouse_metering_history
where warehouse_name=%s
"""
cur.execute(sql, (warehouse_name))

Alle Ergebnisse abrufen

In allen Beispielen in diesem Beitrag haben wir mit fetchall() sämtliche Ergebnisse der Query zurückgegeben:

cur.execute('select * from table')
results=cur.fetchall()

Batches verarbeiten

Wenn Sie den Speicherverbrauch Ihrer Python-Anwendung gering halten möchten, können Sie die Daten in Batches verarbeiten. Hier ein Beispiel, das jeweils 100.000 Datensätze auf einmal verarbeitet:

cur.execute('select * from table')
while True:
    rows = cur.fetchmany(100000)
    if not rows:
        break
    for row in rows:
        # Jede Zeile verarbeiten
        print(row)

Eine Liste von Dictionaries zurückgeben

Standardmäßig liefert Snowflake eine Liste von Tupeln zurück, ohne dass erkennbar ist, zu welcher Spalte ein Eintrag im Tupel gehört. Mit folgendem Beispiel erhalten Sie pro Zeile ein Dictionary, dessen Schlüssel der Spaltenname ist:

1sql = """

2	select

3		warehouse_name,

4		credits_used_compute

5	from snowflake.account_usage.warehouse_metering_history

6	limit 1

7"""

8cur.execute(sql)

9results = cur.fetchall()

10result_meta = cur.description

11results = [\
\
12    dict(zip([col.name for col in result_meta], row)) for row in results\
\
13]

14

15# >>> print(results)

Code ausklappen

Mehrere SQL-Statements auf einmal ausführen

An cursor.execute können Sie immer nur ein einzelnes SQL-Statement übergeben. Um mehrere SQL-Statements in einem Rutsch auszuführen, nutzen Sie die Methode execute_string am Connection-Objekt. Sie erhalten dann eine Liste von Cursorn zurück, mit denen Sie – wie in den Beispielen oben – die Ergebnisse weiterverarbeiten können.

sql="""
delete from table where user_id=123;
insert into table values(...);
"""
cursors = conn.execute_string(sql)

# Bei Bedarf Ergebnisse aus jedem Cursor zurückgeben

Session-Parameter setzen

Beim Aufbau der Snowflake-Verbindung können Sie bequem verschiedene Session-Parameter setzen. Typische Beispiele: einen Query-Tag setzen, damit alle Queries dieser Verbindung getaggt sind, oder ein Query-Timeout definieren, damit Queries nach einer bestimmten Zeit automatisch abgebrochen werden.

conn = snowflake.connector.connect(
    user='XXXX',
    password='XXXX',
    account='XXXX',
    session_parameters={
        'QUERY_TAG': 'COST_ANALYTICS',
        'STATEMENT_TIMEOUT_IN_SECONDS': 3600, # Query nach 1 Stunde automatisch abbrechen
    }
)

Sie können auch weitere Parameter übergeben, etwa warehouse, um festzulegen, auf welchem Virtual Warehouse Ihre Queries laufen. Ohne Angabe werden Queries auf dem Standard-Warehouse des Nutzers ausgeführt.

Eine vollständige Liste der Verbindungsparameter finden Sie hier.