Руководство: Извлечение больших объемов Python в записных книжках Fabric

В этом руководстве вы создадите записную книжку Microsoft Fabric для извлечения данных из нескольких семантических моделей Power BI с помощью REST API Execute DAX Queries. Вы десериализуете Arrow IPC ответы в pandas DataFrames, сравниваете и объединяете выходные данные модели, а также постепенно сливаете результаты в таблицу Delta в OneLake.

Этот шаблон предназначен для специалистов по обработке и анализу данных, которые нуждаются в извлечении высокой пропускной способности с низкой нагрузкой на анализ.

Почему этот шаблон работает

По сравнению с извлечением данных с помощью JSON, использование Arrow IPC снижает нагрузку на процессор и память на стороне клиента, так как позволяет избежать повторного синтаксического анализа JSON и материализации объектов. Буферы Arrow можно считывать непосредственно в табличный формат в памяти и преобразовывать в объект pandas с меньшим количеством шагов преобразования.

При постепенном сохранении результирующих наборов в Delta вы также избегаете полной перезаписи таблиц. Этот подход помогает сократить использование единиц емкости (CU), сохраняя актуальность последующих сценариев Direct Lake.

Что вы строите

В одной записной книжке Fabric вы:

Запрос двух семантических моделей с помощью DAX.
Материализует каждый ответ в виде кадра данных pandas.
Сравните или объедините DataFrame.
Поэтапное слияние изменений в таблицу Delta.
Убедитесь, что потребители Direct Lake могут забрать обновленные данные.

Необходимые условия

Рабочая область емкости Fabric или Premium.
По крайней мере две семантические модели, которые необходимо сравнить или объединить.
Разрешения на сборку и чтение для каждой семантической модели.
Записная книжка Fabric, подключенная к lakehouse, где можно создавать и обновлять таблицы Delta.

пакеты Python:

%pip install msal requests pyarrow pandas

Параметры арендатора включены:
- REST API выполнения запросов набора данных.
- Разрешить основным объектам службы использовать API Power BI при использовании аутентификации только с использованием приложений.

поток записной книжки в системе Fabric

Записная книжка выполняет следующие действия:

Получение токена доступа.
Выполните DAX для нескольких семантических моделей.
Десериализация ответов со стрелками в pandas DataFrames.
Нормализовать схемы и сравнить или объединить DataFrames.
Пошаговое объединение результатов в таблицу Delta.
Проверьте доступность данных для использования Direct Lake.

1. Получение токена Entra Id для текущего пользователя

В первой ячейке кода определите целевые объекты семантической модели и получите маркер.

import notebookutils  # available in every Fabric notebook runtime

# Power BI resource URI — must match this exact value
PBI_RESOURCE = "https://analysis.windows.net/powerbi/api"

# Acquire an Entra Id token for the current user (or workspace identity)
# using the notebook's built-in credential provider.
access_token = notebookutils.credentials.getToken(PBI_RESOURCE)
if access_token is None:
    raise RuntimeError(f"Token acquisition failed")

2. Выполнение запросов DAX в семантических моделях

Определите вспомогательную функцию, которая выполняет DAX и возвращает DataFrame pandas из Arrow IPC.

import io
import pandas as pd
import pyarrow as pa

from datetime import datetime, timezone

def execute_dax_to_pandas(workspace_id: str, dataset_id: str, query: str) -> pd.DataFrame:
    url = (
        f"https://api.powerbi.com/v1.0/myorg/groups/{workspace_id}"
        f"/datasets/{dataset_id}/executeDaxQueries"
    )
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json"
    }
    body = {
        "query": query,
        "resultsetRowcountLimit": 500000
    }

    response = requests.post(url, headers=headers, json=body, timeout=180)
    response.raise_for_status()

    reader = pa.ipc.open_stream(io.BytesIO(response.content))
    table = reader.read_all()
    return table.to_pandas()

В следующей ячейке кода выполните DAX-запрос, специфичный для каждой модели, и пометьте его происхождение.

dax_query = """
EVALUATE
SUMMARIZECOLUMNS(
    'Date'[Date],
    'Product'[ProductKey],
    "NetSales", [Net Sales],
    "Units", [Units]
)
"""

models = [
    {
        "name": "YOUR_FIRST_SEMANTIC_MODEL",
        "workspace_id": "YOUR_WORKSPACE_ID_1",
        "dataset_id": "YOUR_DATASET_ID_1"
    },
    {
        "name": "YOUR_SECOND_SEMANTIC_MODEL",
        "workspace_id": "YOUR_WORKSPACE_ID_2",
        "dataset_id": "YOUR_DATASET_ID_2"
    }
]

frames = []
for m in models:
    df = execute_dax_to_pandas(m["workspace_id"], m["dataset_id"], dax_query)
    df["model_name"] = m["name"]
    df["extract_utc"] = datetime.now(timezone.utc)
    frames.append(df)

print(f"Extracted {len(frames)} DataFrames.")

3. Сравнение и объединение DataFrames

Нормализовать ключевые столбцы, а затем сравнить выходные данные модели или объединить их в один аналитический набор.

for i, df in enumerate(frames):
    df["Date"] = pd.to_datetime(df["Date"], utc=True)
    df["ProductKey"] = df["ProductKey"].astype("int64")
    frames[i] = df

combined_df = pd.concat(frames, ignore_index=True)

# Example comparison: variance between models by date and product
comparison_df = (
    combined_df
    .pivot_table(
        index=["Date", "ProductKey"],
        columns="model_name",
        values="NetSales",
        aggfunc="sum"
    )
    .reset_index()
)

if "sales_model" in comparison_df and "inventory_model" in comparison_df:
    comparison_df["netsales_delta"] = (
        comparison_df["sales_model"] - comparison_df["inventory_model"]
    )

display(comparison_df.head(20))

4. Поэтапное слияние с таблицей Delta

Используйте Delta merge, ключевое поле которого основывается на столбцах бизнес-процессов. Этот шаблон обновляет строки и вставляет новые строки без перезаписи полной таблицы.

# In Fabric notebooks, Spark is available by default.
spark_df = spark.createDataFrame(combined_df)
spark_df.createOrReplaceTempView("stg_semantic_extract")

spark.sql("""
CREATE TABLE IF NOT EXISTS lakehouse.analytics.semantic_extract_delta
USING DELTA
AS
SELECT * FROM stg_semantic_extract WHERE 1 = 0
""")

spark.sql("""
MERGE INTO lakehouse.analytics.semantic_extract_delta AS tgt
USING stg_semantic_extract AS src
ON  tgt.Date = src.Date
AND tgt.ProductKey = src.ProductKey
AND tgt.model_name = src.model_name
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
""")

Tip

Для очень больших окон извлечения секционируйте целевую таблицу Delta по дате и обрабатывайте данные в ограниченных сегментах. Этот подход повышает эффективность слияния и помогает контролировать использование CU.

5. Проверка готовности Direct Lake

Убедитесь, что таблица Delta обновлена и доступна для запросов.

spark.sql("""
SELECT model_name, COUNT(*) AS row_count, MAX(extract_utc) AS latest_extract
FROM lakehouse.analytics.semantic_extract_delta
GROUP BY model_name
""").show(truncate=False)

После обновления таблицы Delta семантические модели Direct Lake, ссылающиеся на нее, могут получать новые данные с помощью нормального поведения синхронизации.

Руководство по производительности

Сохраняйте область применения DAX только для обязательных столбцов и строк.
Используйте resultsetRowcountLimit и фильтры DAX для ограничения окон извлечения.
Предпочитайте инкрементные слияния по сравнению с полной перезаписью данных.
Использовать одного клиента MSAL и кэш токенов в рамках сеанса записной книжки.
Предпочитайте сквозную стрелку для извлечения, чтобы избежать затрат на анализ JSON в Python.
Отслеживайте длительность извлечения, размер полезных данных и длительность слияния в виде операционных метрик.

Troubleshooting

401 Неавторизовано: Проверьте арендатора, учетные данные клиента и область.
HTTP 429: добавьте повторные попытки с экспоненциальной задержкой и джиттером.
Смещение схемы между моделями: нормализация имен столбцов и типов данных перед слиянием.
Использование большого объема памяти в pandas: обработка вывода модели пакетами или агрегирование в DAX перед извлечением.

Замечание

Если вызывающий объект имеет недостаточно разрешений, запрос завершается ошибкой, но HTTP-ответ по-прежнему 200 OK. Проверьте текст ответа для получения сведений об ошибке.

Обратная связь

Были ли сведения на этой странице полезными?

Last updated on 2026-05-07