> ## Documentation Index
> Fetch the complete documentation index at: https://private-7c7dfe99-vortex-format.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Импорт данных трассировки из Langfuse Cloud

> Как загрузить данные трассировки LLM, экспортированные из Langfuse Cloud, в ClickHouse Cloud с помощью ClickPipes.

В этом руководстве пошагово описано, как загрузить данные трассировки LLM из [Langfuse Cloud](https://cloud.langfuse.com/) в ClickHouse Cloud для Real-time аналитики.

<h2 id="export-from-langfuse-cloud">
  Экспорт из Langfuse Cloud
</h2>

Чтобы настроить в Langfuse Cloud интеграцию с Blob Storage для экспорта по расписанию, следуйте [этому руководству](https://langfuse.com/docs/api-and-data-platform/features/export-to-blob-storage). Экспорт данных трассировки в объектное хранилище можно запускать не чаще чем раз в `20 minutes` либо по расписанию `hourly`, `daily` или `weekly`. По умолчанию каждый экспорт включает `observations` (дополненные атрибутами трассировок) и `scores`.

<h3 id="exported-file-layout">
  Структура экспортируемых файлов
</h3>

Интеграция Blob Storage в Langfuse Cloud записывает данные в целевой бакет или контейнер хранилища со следующей структурой каталогов:

```text theme={null}
{prefix}{project-id}/
├── observations_v2/   # observations enriched with trace attributes
├── scores/            # scores
└── manifests/         # one JSON file per completed run with export metadata
```

<h2 id="import-into-clickhouse-cloud">
  Импорт в ClickHouse Cloud
</h2>

После настройки интеграции Blob Storage в Langfuse Cloud данные трассировки будут по расписанию экспортироваться в целевой бакет или контейнер хранилища. Чтобы импортировать эти данные в ClickHouse Cloud, воспользуйтесь [ClickPipes](/ru/integrations/clickpipes) и его управляемыми коннекторами к объектным хранилищам:

| Источник данных | Подробности |
| - | - |
| <span style={{display: 'inline-flex', alignItems: 'flex-start'}}><img src="https://mintcdn.com/private-7c7dfe99-vortex-format/021U3WW1STz_H5Tt/images/integrations/logos/amazon_s3_logo.svg?fit=max&auto=format&n=021U3WW1STz_H5Tt&q=85&s=2412b7c371a24b6f62c44664c13b492b" alt="Логотип Amazon S3" className="integration-table-logo" width="24" height="24" data-path="images/integrations/logos/amazon_s3_logo.svg" /><span><strong>Amazon S3</strong><br /><em>и S3-совместимые объектные хранилища</em></span></span> | Интеграция через [S3 ClickPipes](/ru/integrations/clickpipes/object-storage/amazon-s3/overview). |
| <img src="https://mintcdn.com/private-7c7dfe99-vortex-format/021U3WW1STz_H5Tt/images/integrations/logos/gcs.svg?fit=max&auto=format&n=021U3WW1STz_H5Tt&q=85&s=414fbc807c7c25af0a899441eafb378c" alt="Логотип Google Cloud Storage" className="integration-table-logo" width="24" height="24" data-path="images/integrations/logos/gcs.svg" /> **Google Cloud Storage** | Интеграция через [GCS ClickPipes](/ru/integrations/clickpipes/object-storage/google-cloud-storage/overview). |
| <img src="https://mintcdn.com/private-7c7dfe99-vortex-format/021U3WW1STz_H5Tt/images/integrations/logos/azureblobstorage.svg?fit=max&auto=format&n=021U3WW1STz_H5Tt&q=85&s=aa5acef01b6cc01f2e1c665a8c443b5f" alt="Логотип Azure Blob Storage" className="integration-table-logo" width="24" height="24" data-path="images/integrations/logos/azureblobstorage.svg" /> **Azure Blob Storage** | Интеграция через [ABS ClickPipes](/ru/integrations/clickpipes/object-storage/azure-blob-storage/overview). |

<h3 id="create-a-clickpipe">
  Создание ClickPipe
</h3>

В этом примере S3 ClickPipe используется для импорта данных из S3 бакета, в который по расписанию выгружаются данные из Langfuse Cloud. Этот пример можно взять за основу и для других объектных хранилищ, в которые выполняется экспорт. Рекомендации для конкретных источников данных см. в [документации ClickPipes](/ru/integrations/clickpipes).

<Steps>
  <Step title="Выберите источник данных" id="1-select-the-data-source">
    **1.** В ClickHouse Cloud выберите **Data sources** в главном меню навигации и нажмите **Create ClickPipe**.

    <Frame>
      <img src="https://mintcdn.com/private-7c7dfe99-vortex-format/Kj4wT80mHNH7Aczn/images/integrations/data-ingestion/clickpipes/cp_step0.webp?fit=max&auto=format&n=Kj4wT80mHNH7Aczn&q=85&s=275437c400afd56e6808a20a266c9d7a" alt="Создание ClickPipe" width="2606" height="790" data-path="images/integrations/data-ingestion/clickpipes/cp_step0.webp" />
    </Frame>

    **2.** Нажмите на плитку **Amazon S3**. Через эту плитку можно также подключиться к другим S3-совместимым сервисам, которых нет в интерфейсе ClickPipes.

    <Frame>
      <img src="https://mintcdn.com/private-7c7dfe99-vortex-format/Kj4wT80mHNH7Aczn/images/integrations/data-ingestion/clickpipes/object-storage/amazon-s3/cp_step1.webp?fit=max&auto=format&n=Kj4wT80mHNH7Aczn&q=85&s=4d68698691f1f71f015337e8675c2f08" alt="Выбор источника данных Amazon S3" width="3016" height="954" data-path="images/integrations/data-ingestion/clickpipes/object-storage/amazon-s3/cp_step1.webp" />
    </Frame>
  </Step>

  <Step title="Настройте соединение" id="2-set-up-the-connection">
    **1.** Укажите данные, которые нужны ClickPipes для подключения и аутентификации в бакете, куда экспортируются данные из Langfuse Cloud.

    * **Authentication method**: S3 ClickPipe поддерживает [учётные данные IAM](/ru/integrations/clickpipes/object-storage/amazon-s3/overview#iam-credentials) (`Credentials`) и [аутентификацию на основе роли IAM](/ru/integrations/clickpipes/object-storage/amazon-s3/overview#iam-role) (`IAM role`). ClickPipes требуется только доступ к бакету на чтение. Рекомендации по аутентификации и разрешениям приведены в [справочной документации](/ru/integrations/clickpipes/object-storage/amazon-s3/overview#access-control).

    * **S3 file path**: укажите для ClickPipe подкаталог и используйте подстановочный символ `*`, чтобы охватить все экспортированные файлы. Для каждого подкаталога **необходимо** создать отдельный ClickPipe, поскольку экспортированные файлы в разных подкаталогах имеют разные схемы.

      | Подкаталог | Путь к файлам |
      | - | - |
      | `observations_v2/` | `https://{bucket-name}.s3.{region-code}.amazonaws.com/{prefix}{project-id}/observations_v2/*` |
      | `scores/` | `https://{bucket-name}.s3.{region-code}.amazonaws.com/{prefix}{project-id}/scores/*` |

      Рекомендации по поддерживаемым шаблонам, в том числе по сопоставлению файлов во вложенных префиксах, приведены в [справочной документации](/ru/integrations/clickpipes/object-storage/amazon-s3/overview#file-pattern-matching).

    **2.** Выберите **Continuous ingestion**, чтобы новые файлы загружались автоматически по мере того, как Langfuse записывает их в целевой бакет.

    <Frame>
      <img src="https://mintcdn.com/private-7c7dfe99-vortex-format/7iVRKgAm8eA5yLU2/images/use-cases/AI_ML/Langfuse/cp_continuous_ingestion.webp?fit=max&auto=format&n=7iVRKgAm8eA5yLU2&q=85&s=998932c0a30c9e9d4abc6300b8d51aaa" alt="Сведения о подключении для экспорта Langfuse с включённой непрерывной ингестией." width="3008" height="1542" data-path="images/use-cases/AI_ML/Langfuse/cp_continuous_ingestion.webp" />
    </Frame>

    **3.** Нажмите **Incoming data**. ClickPipes получит метаданные из вашего бакета и на следующем шаге автоматически определит схему целевой таблицы.
  </Step>

  <Step title="Проверьте корректность поступающих данных" id="3-validate-incoming-data">
    **1.** ClickPipes подключится к вашему бакету, получит список файлов по указанному пути и автоматически определит их формат. В этом примере используется формат экспорта Langfuse Cloud по умолчанию (Parquet).

    <Frame>
      <img src="https://mintcdn.com/private-7c7dfe99-vortex-format/7iVRKgAm8eA5yLU2/images/use-cases/AI_ML/Langfuse/cp_incoming_data.webp?fit=max&auto=format&n=7iVRKgAm8eA5yLU2&q=85&s=069e3da21b9294a9e2648300d7458eaf" alt="Файлы, соответствующие пути экспорта Langfuse, с типом файла Parquet" width="3014" height="1574" data-path="images/use-cases/AI_ML/Langfuse/cp_incoming_data.webp" />
    </Frame>

    **2.** Нажмите **Parse information**. ClickPipes определит схему экспортированных данных по файлу-образцу и автоматически сопоставит поля источника с полями целевой таблицы ClickHouse.
  </Step>

  <Step title="Настройте целевую систему" id="4-configure-target">
    **1.** На этом шаге можно просмотреть автоматически определённую схему и настроить параметры целевой таблицы: скорректировать сопоставление типов данных, задать [ключ сортировки](/ru/best-practices/choosing-a-primary-key) и выбрать [движок таблицы](/ru/reference/engines/table-engines).

    <Frame>
      <img src="https://mintcdn.com/private-7c7dfe99-vortex-format/7iVRKgAm8eA5yLU2/images/use-cases/AI_ML/Langfuse/cp_table_settings.webp?fit=max&auto=format&n=7iVRKgAm8eA5yLU2&q=85&s=54708b028d2e7a870f2883da5165a7e9" alt="Экспорт scores с записью в отдельную базу данных langfuse; ключ сортировки задан как environment, name, timestamp, trace_id, observation_id и id" width="3014" height="1578" data-path="images/use-cases/AI_ML/Langfuse/cp_table_settings.webp" />
    </Frame>

    В разделе **Upload data to** оставьте выбранным вариант **New table** и укажите:

    * **Database** и **Name**: отдельную базу данных для данных Langfuse (например, `langfuse`) и понятное имя целевой таблицы (например, `scores`). По умолчанию ClickPipes использует `default.s3-<uuid>`, чтобы избежать конфликтов имён.

    * **Sorting key**: [ключ упорядочивания](/ru/best-practices/choosing-a-primary-key) целевой таблицы, который определяет, как ClickHouse хранит данные на диске. Для оптимальной производительности ключ сортировки должен соответствовать тому, как ваши запросы обращаются к данным, — тогда они смогут пропускать чтение как можно большего объёма данных.

      | Подкаталог | Целевая таблица | Рекомендуемый ключ сортировки |
      | - | - | - |
      | `observations_v2/` | `observations_v2` | `environment, start_time, trace_id, id` |
      | `scores/` | `scores` | `environment, name, timestamp, trace_id, observation_id, id` |

      Этот ключ также используется для дедупликации в таблицах `ReplacingMergeTree`, поэтому набор столбцов должен однозначно идентифицировать запись. Если вы загружаете **несколько проектов Langfuse** в одну таблицу, добавьте `project_id` в начало ключа сортировки. Рекомендации по выбору ключа сортировки см. в [справочной документации](/ru/best-practices/choosing-a-primary-key).

    * **Partition by**: оставьте поле пустым, если не планируете удалять устаревшие трассировки. [Партиционирование](/ru/best-practices/choosing-a-partitioning-key) в ClickHouse предназначено для управления данными, а не для оптимизации запросов; рекомендуемый ключ сортировки и так отсекает лишние данные в запросах по временному диапазону. Если же нужно ограничить срок хранения, партиционируйте таблицу по столбцу времени (`toYYYYMM(start_time)` или `toYYYYMM(timestamp)` для `scores`) — тогда удаление трассировок за месяц сведётся к одной операции `DROP PARTITION`.

    **2.** Затем разверните **Advanced settings**, выберите для **Engine** значение [`ReplacingMergeTree`](/ru/engines/table-engines/mergetree-family/replacingmergetree), чтобы обеспечить дедупликацию, и укажите `updated_at` в качестве столбца **Version**.

    <Frame>
      <img src="https://mintcdn.com/private-7c7dfe99-vortex-format/7iVRKgAm8eA5yLU2/images/use-cases/AI_ML/Langfuse/cp_advanced_settings.webp?fit=max&auto=format&n=7iVRKgAm8eA5yLU2&q=85&s=db83f02c195266677613b8036b347421" alt="Расширенные настройки с движком ReplacingMergeTree и updated_at в качестве столбца версии" width="2400" height="484" data-path="images/use-cases/AI_ML/Langfuse/cp_advanced_settings.webp" />
    </Frame>

    Окна экспорта в интеграции Langfuse Cloud Blob Storage включают обе границы, поэтому одна и та же запись может попасть в несколько экспортированных файлов. Если ваши аналитические запросы не идемпотентны (идемпотентны, например, `uniq()`, `max()`, `min()`), вы **обязаны** обрабатывать дедупликацию на уровне последующей модели данных, используя `ReplacingMergeTree` в качестве целевого движка.

    <Note>
      **Дедупликация с помощью ReplacingMergeTree**

      Этот движок таблицы схлопывает дубликаты во время фоновых слияний. Для полной дедупликации при чтении (т. е. семантики merge-on-read) **необходимо** использовать в запросах модификатор `FINAL` (или агрегации в стиле `argMax()`).
    </Note>
  </Step>

  <Step title="Настройте разрешения" id="5-configure-permissions">
    ClickPipes создаёт отдельного пользователя для записи данных в целевую таблицу. Выберите для этого внутреннего пользователя пользовательскую роль или одну из предопределённых ролей:

    * `Full access`: полный доступ к кластеру. Требуется, если с целевой таблицей используется materialized view или словарь.
    * `Only destination table`: разрешения `INSERT` только на целевую таблицу.

    <Frame>
      <img src="https://mintcdn.com/private-7c7dfe99-vortex-format/Kj4wT80mHNH7Aczn/images/integrations/data-ingestion/clickpipes/cp_step5.webp?fit=max&auto=format&n=Kj4wT80mHNH7Aczn&q=85&s=ac9fe916e6c9fe09445c8b9fe0f3934f" alt="Разрешения" width="2736" height="1318" data-path="images/integrations/data-ingestion/clickpipes/cp_step5.webp" />
    </Frame>

    Нажмите **Complete setup**, чтобы создать ClickPipe.
  </Step>

  <Step title="Завершите настройку" id="6-complete-setup">
    Готово! Сначала ClickPipes выполнит дозагрузку исторических данных из всех файлов по указанному пути, а затем начнёт принимать новые файлы по мере их поступления в бакет.

    <Frame>
      <img src="https://mintcdn.com/private-7c7dfe99-vortex-format/7iVRKgAm8eA5yLU2/images/use-cases/AI_ML/Langfuse/cp_clickpipes_running.webp?fit=max&auto=format&n=7iVRKgAm8eA5yLU2&q=85&s=707127f711479efe0dbc8892bd4cb4f7" alt="Работающие ClickPipes для observations и scores в разделе Data sources" width="3020" height="874" data-path="images/use-cases/AI_ML/Langfuse/cp_clickpipes_running.webp" />
    </Frame>

    Повторите шаги из этого руководства для каждого подкаталога, который нужно импортировать в ClickHouse Cloud.
  </Step>
</Steps>

<h3 id="query-trace-data">
  Запросы к данным трассировки
</h3>

Когда ClickPipes будут запущены, вы сможете выполнять запросы к данным трассировки Langfuse Cloud непосредственно в ClickHouse Cloud. Ниже приведены примеры нескольких распространённых шаблонов запросов: агрегирование затрат и задержек, восстановление отдельной трассировки, а также сопоставление оценок с моделями и промптами.

<h4 id="time-bucketed-metric-rollups">
  Агрегация метрик по временным интервалам
</h4>

<Tip>
  Некоторые поля, например `usage_details` и `cost_details`, экспортируются как столбцы типа `Map`. Чтобы получить отдельное значение, обратитесь к нему по ключу (например, `sum(cost_details['total'])`).
</Tip>

```sql theme={null}
SELECT
    toStartOfHour(start_time) AS t,
    provided_model_name,
    count()                        AS spans,
    quantile(0.95)(latency)        AS p95_latency,
    sum(total_cost)                AS cost,
    sum(usage_details['total'])    AS tokens
FROM observations_v2 FINAL
WHERE start_time >= now() - INTERVAL 7 DAY
-- AND environment = '<environment>'
GROUP BY t, provided_model_name
ORDER BY t;
```

<h4 id="trace-reconstruction">
  Восстановление трассировки
</h4>

```sql theme={null}
SELECT *
FROM observations_v2 FINAL
WHERE trace_id = '<trace_id>'
ORDER BY start_time;
```

<h4 id="quality-analysis">
  Анализ качества
</h4>

```sql theme={null}
SELECT
    o.provided_model_name,
    s.name,
    avg(s.value) AS avg_score
FROM "s3-26237f63-84dc-4a7d-a0e1-7361cf55ee0b" AS s FINAL
JOIN "s3-80899034-ce1e-4dc9-a4e6-b059ddec4175" AS o FINAL ON s.trace_id = o.trace_id
--WHERE s.environment = '<environment>'
GROUP BY o.provided_model_name, s.name
ORDER BY avg_score DESC;
```

<h2 id="see-also">
  См. также
</h2>

* [Langfuse и ClickHouse](/ru/products/cloud/features/ai-ml/langfuse)
* [Экспорт данных Langfuse Cloud в Blob Storage](https://langfuse.com/docs/api-and-data-platform/features/export-to-blob-storage)
* [ClickPipes для объектного хранилища](/ru/integrations/clickpipes/object-storage/amazon-s3/overview)