An eine Cloud-Datenplattform wie Snowflake lassen sich über ELT-Strecken verschiedenste Quellsysteme anbinden. Die Anbindung erfolgt, abhängig vom Quellsystem, auf unterschiedliche Art und Weise. Eine gängige Methode ist die Anbindung über eine API, da viele Systeme diese als Datenschnittstelle anbieten. Daten aus einer API in ein Data Warehouse zu laden, gehört daher zu den typischen Aufgaben von DWH-Entwickler:innen. In der Dokumentation von Snowflake gibt es zu dem Thema leider wenig zu finden. Wir wollen daher an einem Beispiel zeigen, wie man mit Snowflake-Mitteln eine API ansprechen kann und anschließend die Daten transformieren und in Tabellen laden kann.
Beispielszenario
In der viadee haben wir vor kurzem ein internes Snowflake-Demo-DWH aufgebaut, welches wir für Schulungen und Tests von Tools nutzen. Das DWH enthält Daten aus einem Dorfladen, der Lebensmittel und Produkte des täglichen Bedarfs verkauft. Das Kassensystem des Dorfladens bietet die Möglichkeit, per API auf die Tabellen “Kunden”, “Wareneingänge”, “Zulieferer” etc. zuzugreifen. Über eine ELT-Strecke rufen wir täglich die API auf und erhalten die tagesaktuellen Daten als JSON und laden diese in das DWH. Den Aufbau dieser ELT-Strecke werden wir im Folgenden anhand der Zulieferer (Supplier) aufzeigen. Die Struktur der JSON-Daten für die Zulieferer ist hier beispielhaft dargestellt:
{
"currentPage": 1,
"links": {
"self": "/web/api/v3/accounts/account-ID/suppliers?page=1"
},
"maxRevision": 183,
"pagesTotal": 1,
"results": [
{
"active": true,
"address": {
"address": {
"addressLine1": "Hauptstraße 23",
"city": "Münster",
"company": "Gemüsehandel Mustermann",
"zipCode": "48147"
}
},
"id": "0400c6f5-03af-4118-93ba-73d9a3683ae2",
"name": "Gemüsehandel Mustermann",
"number": "693010"
},
{
"active": true,
"address": {
"address": {
"addressLine1": "Landstraße 56",
"city": "Münster",
"company": "Milchbauer Meier",
"zipCode": "48147"
}
},
"id": "0bcc7c49-7596-4d3b-bf77-bb84911b6f5f",
"name": "Milchbauer Meier",
"number": "694004"
}
],
"resultsOfPage": 2,
"resultsTotal": 2
}Wie funktioniert der Zugriff auf die API von Snowflake aus?
Um von Snowflake aus den Zugriff auf eine API zu ermöglichen, müssen zunächst einige Elemente angelegt werden:
Network Rule:
Eine Network Rule regelt, welche Zugriffe von außen nach Snowflake (ingress) oder von Snowflake nach außen (egress) erlaubt sind. Standardmäßig sind Zugriffe nach außen erstmal nicht möglich. Es muss daher eine Network Rule angelegt werden, die den Zugriff auf die URL der API erlaubt.Secrets/Security-Integration (optional):
Secrets können genutzt werden, um sicher Benutzernamen und Passwörter zu speichern, um sich bei der API zu authentifizieren. Eine Security Integration ermöglicht auch eine OAuth-Authentifizierung bei der API. Wenn die API nicht gesichert ist, werden diese Elemente nicht benötigt.External-Access-Integration:
Eine External-Access-Integration dient dazu, die Network-Rule und ggf. die Secrets für Stored-Procedures und User-Defined-Functions (UDFs) verfügbar zu machen. Diese kann nur von einem Admin angelegt werden.UDF / Stored-Procedure:
Die UDF bzw. die Stored-Procedure enthält den Code für den Zugriff auf die API und das Laden der Daten nach Snowflake.
1. Erstellen der Network Rule
Um den Zugriff auf die API in Snowflake zu erlauben, müssen wir als Erstes eine Network Rule erstellen. Bei uns sind die Network Rules in der SECURITYDB in einem eigens dafür vorgesehenen Schema gespeichert:
USE ROLE SECURITYADMIN;
CREATE NETWORK RULE SECURITYDB.NETWORK_RULES.DORFLADEN_API_RULE
TYPE = HOST_PORT
MODE = EGRESS
VALUE_LIST = ('api.kassensystem.com');Die Network Rule muss vom Type HOST_PORT sein, damit Netzwerkverkehr basierend auf einer Domain erlaubt werden kann. Die Domain der API wird dann in der Value-List eingetragen. Als Mode muss EGRESS angegeben werden, da es sich um einen Zugriff von Snowflake nach außen handelt.
2. Erstellen der Secrets
In unserem Fall ist der Zugriff auf die API über Benutzername und Passwort gesichert. Daher legen wir in einem separaten Schema ein Secret dafür an. Zusätzlich wird noch ein weiteres Secret erstellt, welches die Account-ID des Kassensystems beinhaltet, da diese nicht für alle Nutzer:innen sichtbar sein soll. Die Secrets sollten unbedingt als Admin angelegt werden und die Nutzung sollte nur gezielten Rollen erlaubt werden! Um Secrets in einer UDF/Stored-Procedure nutzen zu können, muss die Rolle, die diese Prozedur anlegt, die Rechte USAGE und READ auf den Secrets haben. Wir legen daher die Secrets mit der Rolle SECURITYADMIN an und geben die Rechte zur Nutzung an die Rolle DORFLADEN_ADMIN, welche vom technischen User genutzt wird.
USE ROLE SECURITYADMIN;
CREATE SECRET SECURITYDB.SECRETS.DORFLADEN_API_AUTHENTICATION
TYPE = PASSWORD
USERNAME = '*******'
PASSWORD = '*******'
COMMENT = 'Zugangsdaten für die Dorfladen API';
GRANT USAGE ON SECRET SECURITYDB.SECRETS.DORFLADEN_API_AUTHENTICATION TO ROLE DORFLADEN_ADMIN;
GRANT READ ON SECRET SECURITYDB.SECRETS.DORFLADEN_API_AUTHENTICATION TO ROLE DORFLADEN_ADMIN;
CREATE SECRET SECURITYDB.SECRETS.DORFLADEN_ACCOUNT
TYPE = GENERIC_STRING
SECRET_STRING = '*******'
COMMENT = 'Account ID des Kassensystems';
GRANT USAGE ON SECRET SECURITYDB.SECRETS.DORFLADEN_ACCOUNT TO ROLE DORFLADEN_ADMIN;
GRANT READ ON SECRET SECURITYDB.SECRETS.DORFLADEN_ACCOUNT TO ROLE DORFLADEN_ADMIN;3. Erstellen der External-Access-Integration
Als Nächstes benötigen wir eine External-Access-Integration, welche die Network Rule und die Secrets für unsere Stored-Procedure zugreifbar macht. Die External Access Integration stellt also die Brücke zwischen den Elementen dar. Somit kann für jede UDF bzw. Stored-Procedure gesteuert werden, auf welche externen Adressen der Code zugreifen darf und welche Secrets er zur Authentifizierung nutzen kann. Standardmäßig kann eine External-Access-Integration nur von einem Accountadmin angelegt werden. Dieses Recht kann auch an andere Rollen gegeben werden. Es ist allerdings sinnvoll, diese Elemente nur von Admins anlegen zu lassen und anschließend die Nutzung des Elements anderen Rollen zu gewähren.
USE ROLE ACCOUNTADMIN;
CREATE EXTERNAL ACCESS INTEGRATION DORFLADEN_ACCESS_INTEGRATION
ALLOWED_NETWORK_RULES = (SECURITYDB.NETWORK_RULES.DORFLADEN_API_RULE)
ALLOWED_AUTHENTICATION_SECRETS = (SECURITYDB.SECRETS.DORFLADEN_API_AUTHENTICATION, SECURITYDB.SECRETS.DORFLADEN_ACCOUNT)
ENABLED = TRUE;
GRANT USAGE ON INTEGRATION DORFLADEN_ACCESS_INTEGRATION to ROLE DORFLADEN_ADMIN;4. Erstellen einer Stored-Procedure zum Laden der JSON-Daten aus der API
Letztlich erstellen wir eine Stored-Procedure in Python, welche den Code für den Zugriff auf die Daten und das Laden nach Snowflake beinhaltet. In den Parametern müssen sowohl die Secrets als auch die External-Access-Integration angegeben werden. Sollte hier in der Prozedur ein Secret angegeben werden, welches nicht Teil von einer der External-Access-Integrations ist, kommt es zu einem Fehler. Über den Parameter IMPORTS wird eine zusätzliche Python-Datei aus einer Snowflake Stage mit dem Namen “PYTHON_LIB_STAGE” importiert. In dieser Datei befindet sich der Code zum Zugriff auf eine API inklusive Fehlerhandling. Da dieser Code einheitlich von allen Prozeduren zum Laden genutzt werden kann, wurde dieser zentral dort abgelegt und hier nur die Funktion fetch_api_data importiert, welche JSON-Daten als Ergebnis zurück gibt.
Zu Beginn der Funktion fetchData werden die Secrets ausgelesen und in Variablen geschrieben. Dies geschieht über die Funktionen get_username_password bzw. get_generic_secret_string, welche über das Snowflake-interne Hilfsmodul _snowflake aufgerufen werden können. Anschließend wird die URL zusammengebaut und der Aufruf der API gestartet. In der Schleife wird dann die jeweilige Seite mit den JSON-Daten in ein Pandas-Data-Frame geladen und anschließend über die Snowflake-Funktion write_pandas in eine Tabelle geschrieben. Dabei werden die JSON-Daten in die Spalte “JSON_DATA“ vom Typ Variant geschrieben.
CREATE OR REPLACE PROCEDURE RAW.fetchSuppliers(database_name VARCHAR)
RETURNS VARCHAR
LANGUAGE PYTHON
RUNTIME_VERSION = '3.13'
PACKAGES = ('requests', 'pandas', 'snowflake-snowpark-python')
IMPORTS = ('@python_lib_stage/fetch_api_data.py')
HANDLER = 'fetchData'
EXTERNAL_ACCESS_INTEGRATIONS = (DORFLADEN_ACCESS_INTEGRATION)
SECRETS = ('api_authentication' = SECURITYDB.SECRETS.DORFLADEN_API_AUTHENTICATION,
'account' = SECURITYDB.SECRETS.DORFLADEN_ACCOUNT)
EXECUTE AS CALLER
AS
$$
from fetch_api_data import fetch_api_data
import pandas as pd
import _snowflake
def fetchData(snowflake_session, database_name: str):
api_authentication = _snowflake.get_username_password('api_authentication')
api_password = api_authentication.password
api_user = api_authentication.username
account = _snowflake.get_generic_secret_string('account')
apiurl=f'https://api.kassensystem.com/web/accounts/{account}/suppliers'
response = fetch_api_data(apiurl, auth_value=(api_user, api_password))
num_pages = response['pagesTotal']
for page in range(1, num_pages + 1):
response_page = fetch_api_data(apiurl, auth_value=(api_user, api_password), params={'page': page})
df = pd.DataFrame([{"JSON_DATA": response_page, "LOAD_TYPE": 'FULL-LOAD'}])
snowflake_session.write_pandas(df=df, table_name='SUPPLIERS_JSON', database=f'{database_name}', schema='RAW', overwrite=False)
return 'Data fetched successfully!'
$$;5. Erstellen einer Stored-Procedure zum Überführen der JSON-Daten in eine strukturierte Tabelle
Nun haben wir zumindest die JSON-Daten in eine Snowflake Tabelle geschrieben. Allerdings sind die Daten als ein einzelner, großer JSON-String in einer Spalte vom Typ Variant. Was jetzt noch fehlt, ist die Überführung der Daten vom semistrukturierten Format in eine strukturierte Tabelle. Dazu haben wir eine weitere Stored-Procedure in SQL geschrieben, welche den JSON-String parst und per Lateral Flatten das enthaltene Array “results“ in Zeilen aufteilt. Zusätzlich wird noch ein Hash-Wert gebildet, der für die spätere Historisierung der Daten genutzt wird.
CREATE OR REPLACE PROCEDURE RAW.loadSuppliers()
RETURNS VARCHAR
LANGUAGE SQL
EXECUTE AS CALLER
AS
$$
BEGIN
CREATE OR REPLACE TABLE RAW.SUPPLIERS(
SUPPLIER_ID VARCHAR,
SUPPLIER_NUMBER NUMBER,
SUPPLIER_NAME VARCHAR,
ADDRESS VARCHAR,
ZIP_CODE NUMBER,
CITY VARCHAR,
COMPANY VARCHAR,
ACTIVE BOOLEAN,
HASH_VALUE NUMBER(19,0),
LOADED_AT TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP
) as
Select Flatten_Results.VALUE:id::varchar as SUPPLIER_ID,
Flatten_Results.VALUE:"number"::number as SUPPLIER_NUMBER,
Flatten_Results.VALUE:name::varchar as SUPPLIER_NAME,
Flatten_Results.VALUE:address:address:addressLine1::varchar as ADDRESS,
Flatten_Results.VALUE:address:address:zipCode::number as ZIP_CODE,
Flatten_Results.VALUE:address:address:city::varchar as CITY,
Flatten_Results.VALUE:address:address:company::varchar as COMPANY,
Flatten_Results.VALUE:active::boolean as ACTIVE,
HASH(
SUPPLIER_ID,
SUPPLIER_NUMBER,
SUPPLIER_NAME,
ADDRESS,
ZIP_CODE,
CITY,
COMPANY,
ACTIVE
) as HASH_VALUE,
LOADED_AT
FROM RAW.SUPPLIERS_JSON,
LATERAL FLATTEN (input => JSON_DATA:results) as Flatten_Results
;
RETURN 'Suppliers loaded successfully!';
END;
$$
;Die beiden Stored-Procedures werden in der täglichen Verarbeitungspipeline nacheinander ausgeführt. Somit werden die Daten jeden Tag aus der API geladen und in eine strukturierte Tabelle in Snowflake überführt. Die Tabelle wird dabei jedes Mal überschrieben und enthält den aktuellen Stand.
Fazit
Wir haben gezeigt, wie man mit geringem Aufwand und nur mit Snowflake-Mitteln Daten aus einer API in eine Tabelle laden kann. Dazu haben wir in Snowflake eine Network Rule und Secrets angelegt, die im Zusammenspiel den Zugriff nach außen ermöglichen und eine sichere Authentifizierung an der API ermöglichen. Über die External-Access-Integration haben wir die Elemente für die Stored-Procedure verfügbar gemacht, die die API aufruft und die Daten in Snowflake speichert. Der Zusammenhang der erstellten Elemente ist in diesem Diagramm noch einmal veranschaulicht.
Nachdem die Daten ins DWH geladen wurden, können diese auf gewohnte Art und Weise weiterverarbeitet werden. In unserem Demo-DWH geschieht die weitere Transformation über ein DBT-Projekt, welches nativ in Snowflake läuft. Dazu folgt in Kürze ein weiterer Blogartikel.
Wenn Sie Interesse haben ein Data Warehouse aufzubauen, ob in Snowflake oder On-Premise, sprechen Sie uns gerne an.
Wir bieten auch Schulung in den Themenfeldern Grundlagen von Data Warehousing und Cloud Data Warehouse an
