Jak načíst data do Fabric Lakehouse Pythonem, aniž zabijete F2 kapacitu

Vstupním bodem do Microsoft Fabricu, abyste mohli analyzovat data v Power BI, je Lakehouse. Něco mezi datovým skladem a Data Lake, který si bere to nejlepší z obou světů. Nejste svázáni přílišnou strukturovaností jako u Data Lake, ale zároveň máte k dispozici SQL Analytics Endpoint pro dotazování nad Delta Lake tabulkami. Tématu Lakehouse jsem se věnoval v jiných přednáškách a v podcastu Cesta do Fabricu s Vojtou Šímou.

Nyní se zaměříme na to, jak data do Lakehouse dostat. Máme čtyři způsoby a tento článek nebude pokrývat všechny.

• Shortcuts (ala zástupce na ploše)
• Pipelines (ala Azure Data Factory)
• Dataflows Gen 2 (Power Query Low Code)
• Notebooky v Pythonu

Dnes se zaměřím na poslední zmíněné – notebooky v Pythonu. Pro uživatele přicházející z Power BI světa možná nejméně intuitivní, ale mohou být efektivní, co se týče CU (Capacity Units), pokud je používáte dobře.

Porovnáme si čtyři různé typy knihoven v Pythonu, které mi umožní načíst data do Lakehouse. Na paškál jsem si vzal demo dataset New York City Taxi data, který je rozumně velký – necelé dva roky dat (01/2024–09/2025), celkem 76 milionů záznamů.

PySpark

Defaultně vám Fabric nabídne PySpark, který startuje Spark cluster a má několik pracujících uzlů. Jinými slovy – na malé kapacitě jdete okamžitě do burstingu, krátkodobě přetížíte kapacitu a jedete na dluh, kdy si půjčujete Capacity Units z budoucnosti.

Kód by mohl vypadat následovně. Aby si s Timestamp daty poradil i SQL Analytics Endpoint, můžete si všimnout CASTů.

import psutil, os

from pyspark.sql import functions as F

from pyspark.sql.types import TimestampType

def print_memory(label):

    mem = psutil.Process(os.getpid()).memory_info().rss / 1024**3

    print(f"[{label}] RAM used: {mem:.2f} GB")

print_memory("start")

spark.conf.set("spark.sql.parquet.outputTimestampType", "TIMESTAMP_MICROS")

df = spark.read.parquet("Files/NYC/")

print_memory("after read")

df = df.withColumn("tpep_pickup_datetime", F.col("tpep_pickup_datetime").cast(TimestampType())) \

       .withColumn("tpep_dropoff_datetime", F.col("tpep_dropoff_datetime").cast(TimestampType()))

df.write.format("delta") \

    .mode("overwrite") \

    .option("overwriteSchema", "true") \

    .saveAsTable("yellow_tripdata_spark")

print_memory("after write")

Doba trvání na F2: 1:23. Pro menší objemy dat čekáte zbytečně na start clusteru, abyste zapsali pár řádků, a přetížíte Fabric F2 kapacitu. Důležité je, že to funguje.

Pandas

Pokud ale válčíme na malé kapacitě s burstingem, mohli bychom zkusit čistý Python. Jedna z nejpoužívanějších knihoven pro práci s daty v Pythonu je Pandas. Sám, když jsem s Pythonem začínal, byla Pandas jedna z prvních knihoven v rámci kurzů.

Kód by mohl vypadat například následovně:

import pandas as pd

import psutil

import os

from deltalake.writer import write_deltalake

def print_memory(label):

    mem = psutil.Process(os.getpid()).memory_info().rss / 1024**3

    print(f"[{label}] RAM used: {mem:.2f} GB")

print_memory("start")

# Pandas loads ALL files into RAM at once

df = pd.read_parquet('/lakehouse/default/Files/NYC/')

print_memory("after read")  # likely crashes here or reports 2-3 GB

write_deltalake(

    lakehouse.properties['abfsPath'] + "/Tables/yellow_tripdata_pandas",

    df,

    mode="overwrite",

    engine="pyarrow"

)

print_memory("after write")

Nebudu vás napínat – Pandas se snaží načíst všechna data do paměti, než je začne zapisovat. Na malé kapacitě, jako je F2, jsem při těchto objemech dat opakovaně skončil na Out Of Memory Exception. Trvalo to cca 1:45 – pokud na tom záleží, když to nedoběhlo. Odpočívej v pokoji, Pando, alespoň pro use case větší objem dat na malé kapacitě.

DuckDB

Když jsem se před nějakou dobou bavil se Štěpánem Rešlem (kterého tímto zdravím, pokud tohle náhodou čte), doporučil mi knihovny DuckDB a Polars jako efektivnější při využití paměti s podporou stránkování. Tak jsem to zkusil.

Stránkování jsem nastavil na 100 000 záznamů. Kód notebooku vypadá následovně:

import duckdb

from deltalake.writer import write_deltalake

import notebookutils

lakehouse = notebookutils.lakehouse.get("NYC_taxi_data")

token = notebookutils.credentials.getToken("storage")

con = duckdb.connect()

# Cast timestamps explicitly to TIMESTAMP (no timezone, microsecond precision)

reader = con.execute("""

    SELECT

        * EXCLUDE (tpep_pickup_datetime, tpep_dropoff_datetime),

        tpep_pickup_datetime::TIMESTAMPTZ AS tpep_pickup_datetime,

        tpep_dropoff_datetime::TIMESTAMPTZ AS tpep_dropoff_datetime

    FROM read_parquet('/lakehouse/default/Files/NYC/*.parquet')

""").fetch_record_batch(rows_per_batch=100_000)

write_deltalake(

    lakehouse.properties['abfsPath'] + "/Tables/yellow_tripdata_duckDb",  # match exact casing

    reader,

    mode="overwrite",

    engine="rust",

    storage_options={"bearer_token": token, "use_fabric_endpoint": "true"}

)

Musel jsem opět řešit timestamp sloupce, stejně jako u PySparku. Díky stránkování F2 nepřekročila svých dedikovaných 16 GB paměti a load doběhl za 2:20. O něco pomalejší než PySpark, ale single node a kachna nevylezla z paměti a udělala to, co bylo potřeba.

Polars

Díky doporučení od Štěpána a šťourání od Vojty Šímy (kterého taky zdravím 🙂) jsem si řekl, že zkusím totéž i v Polars. Tady to bylo na nejvíce iterací, ale nakonec jsem se dostal k funkčnímu kódu, který by mohl vypadat takhle:

import pyarrow.dataset as ds

import pyarrow as pa

from deltalake.writer import write_deltalake

import notebookutils

import psutil, os, time

lakehouse = notebookutils.lakehouse.get("NYC_taxi_data")

token = notebookutils.credentials.getToken("storage")

def print_memory(label):

    mem = psutil.Process(os.getpid()).memory_info().rss / 1024**3

    print(f"[{label}] RAM used: {mem:.2f} GB")

start = time.time()

print_memory("start")

dataset = ds.dataset('/lakehouse/default/Files/NYC/', format='parquet')

# Check what type PyArrow sees

ts_field = dataset.schema.field("tpep_pickup_datetime")

print(f"Original type: {ts_field.type}")

# Force timestamps to us with UTC timezone — readable by SQL Analytics Endpoint

target_type = pa.timestamp("us", tz="UTC")

schema = dataset.schema

for col in ["tpep_pickup_datetime", "tpep_dropoff_datetime"]:

    idx = schema.get_field_index(col)

    schema = schema.set(idx, schema.field(col).with_type(target_type))

print_memory("after scan")

write_deltalake(

    lakehouse.properties['abfsPath'] + "/Tables/yellow_tripdata_polars",

    dataset.to_batches(),

    schema=schema,

    mode="overwrite",

    engine="rust",

    storage_options={"bearer_token": token, "use_fabric_endpoint": "true"}

)

print_memory("after write")

print(f"Execution time: {time.time() - start:.1f}s")

Doba trvání: 1:22, doběhla a funguje.

Závěr

Stejné cvičení, různé nástroje. Tři ze čtyř soutěžících došli do cíle. Každý má své silné a slabé stránky – každý ať si udělá závěry sám, nebo provede vlastní měření. Nebudu předstírat, že jsem guru na všechny zmiňované knihovny, a pomáhal mi kamarád Claude.

Osobně si z toho odnáším následující postřehy. PySparku se snažím pokud možno vyhýbat na malé kapacitě – už se mi stalo, že jsem si ji při jiných akcích na 24 hodin odstavil. Pandas žere paměť jako by se jednalo o eukalyptus a může přečerpat limit. S Polarsem mám ve finále hezké rozuzlení a dobrý čas, trvalo mi ale za pomoci LLM nejdéle to rozchodit tak, aby doběhlo. DuckDB funguje, dobíhá a podařilo se mi ji rozchodit velmi rychle i s laděním chyb – i když za cenu o něco delšího běhu.

Nebudu tvrdit, že některá knihovna je špatná nebo lepší než ostatní. To si musí každý zhodnotit sám a záleží primárně na kontextu, čeho se snaží dosáhnout. Já ale budu při příštích podobných use cases pošilhávat po DuckDB a Polars.