Pracując w notatnikach Microsoft Fabric potrzebujemy czasem sięgać po jakąś informację do małych tabel (np. konfiguracyjnych). Uruchamianie do tego procesu Spark nie zawsze jest najlepszym rozwiązaniem, szczególnie w przypadku, gdy zależy nam na czasie. 

Jako oczywiste rozwiązanie nasuwa się dostęp do tych tabel przez SQL Endpoint oraz pyODBC. Jednak czy da się to zrobić bez podawania użytkownika i hasła, korzystając z połączenia, które przecież mamy w notatniku? 

Okazuje się, że tak! A to dzięki pakietowi mssparkutils. 

Załóżmy, że mamy w obszarze roboczym lakehouse o nazwie lh_SQLConnectionTest, a w nim tabelę d_calendar zawierającą kalendarz od 2000-01-01 do 2024-12-31. 

Zasymulujemy częste odpytywanie tej tabeli pytając w pętli o kolejne 30 dni:

from datetime import date, timedelta 
vStart = date(2024,7,1) 
for vDay in [d for d in (vStart + timedelta(n) for n in range(30))]: 
    vSql = "select id from d_calendar where date = '{day}'".format(day = vDay) 
    vRes = spark.sql(vSql).collect()[0][0] 

Jak widać wykonanie tego skryptu trwa ok 17. sekund:

Aby podłączyć się do SQL Endpoint musimy znać jego adres. Możemy go sprawdzić w ustawieniach naszego lakehouse: 

Następnie korzystając z mssparkutils.credentials.getToken, możemy uzyskać token bieżącego połączenia i delikatnie go przerabiając połączyć się do SQL Enpoint przy pomocy pyODBC: 

import struct 
from itertools import chain, repeat 
import pyodbc 
sqlEndpoint = '*****.datawarehouse.fabric.microsoft.com' 
connectionString = f"Driver={{ODBC Driver 18 for SQL Server}};Server={sqlEndpoint},1433;Encrypt=Yes;TrustServerCertificate=No" 
token_object = mssparkutils.credentials.getToken("pbi") 
encoded_bytes = bytes(chain.from_iterable(zip(bytes(token_object, "UTF-8"), repeat(0)))) 
token_bytes = struct.pack("<i", len(encoded_bytes)) + encoded_bytes 
attrs_before = {1256: token_bytes} 
connection = pyodbc.connect(connectionString, attrs_before=attrs_before) 

Wykonanie tego samego zapytania przy użyciu pyODBC może wyglądać następująco:

from datetime import date, timedelta 
cursor = connection.cursor() 
vStart = date(2024,7,1) 
for vDay in [d for d in (vStart + timedelta(n) for n in range(30))]: 
    vSql = "select id from lh_SQLConnectionTest.dbo.d_calendar where date = '{day}'".format(day = vDay) 
    cursor.execute(vSql) 
    vWyn = cursor.fetchall()[0][0] 

Zwróć uwagę na konieczność podania pełnej nazwy tabeli z nazwą lakehouse i schematem. 

Wykonanie tego skryptu trwa już tylko 5 sekund. 

Jeśli chcesz poeksperymentować, wrzuć do swojego Fabrica załączony notatnik (nie zapomnij tylko podpiąć swój lakehouse do notatnika).  

Napisałem tam klasę użytkową obsługującą połączenie SQL Endpoint, którą możesz wykorzystać w swoim projekcie. 

Miłego eksperymentowania 😊 

en_GB