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

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

В этом руководстве пошагово описано, как загрузить данные трассировки LLM из Langfuse Cloud в ClickHouse Cloud для Real-time аналитики.

​

Экспорт из Langfuse Cloud

Чтобы настроить в Langfuse Cloud интеграцию с Blob Storage для экспорта по расписанию, следуйте этому руководству. Экспорт данных трассировки в объектное хранилище можно запускать не чаще чем раз в 20 minutes либо по расписанию hourly, daily или weekly. По умолчанию каждый экспорт включает observations (дополненные атрибутами трассировок) и scores.

​

Структура экспортируемых файлов

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

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

​

Импорт в ClickHouse Cloud

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

Источник данных	Подробности
Amazon S3 и S3-совместимые объектные хранилища	Интеграция через S3 ClickPipes.
Google Cloud Storage	Интеграция через GCS ClickPipes.
Azure Blob Storage	Интеграция через ABS ClickPipes.

​

Создание ClickPipe

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

1. Выберите источник данных

   1. В ClickHouse Cloud выберите Data sources в главном меню навигации и нажмите Create ClickPipe.

   

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

   
2. Настройте соединение

   1. Укажите данные, которые нужны ClickPipes для подключения и аутентификации в бакете, куда экспортируются данные из Langfuse Cloud.

   • Authentication method: S3 ClickPipe поддерживает учётные данные IAM (Credentials) и аутентификацию на основе роли IAM (IAM role). ClickPipes требуется только доступ к бакету на чтение. Рекомендации по аутентификации и разрешениям приведены в справочной документации.

   • 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/*

     Рекомендации по поддерживаемым шаблонам, в том числе по сопоставлению файлов во вложенных префиксах, приведены в справочной документации.

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

   

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

3. Проверьте корректность поступающих данных

   1. ClickPipes подключится к вашему бакету, получит список файлов по указанному пути и автоматически определит их формат. В этом примере используется формат экспорта Langfuse Cloud по умолчанию (Parquet).

   

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

4. Настройте целевую систему

   1. На этом шаге можно просмотреть автоматически определённую схему и настроить параметры целевой таблицы: скорректировать сопоставление типов данных, задать ключ сортировки и выбрать движок таблицы.

   

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

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

   • Sorting 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 в начало ключа сортировки. Рекомендации по выбору ключа сортировки см. в справочной документации.

   • Partition by: оставьте поле пустым, если не планируете удалять устаревшие трассировки. Партиционирование в ClickHouse предназначено для управления данными, а не для оптимизации запросов; рекомендуемый ключ сортировки и так отсекает лишние данные в запросах по временному диапазону. Если же нужно ограничить срок хранения, партиционируйте таблицу по столбцу времени (toYYYYMM(start_time) или toYYYYMM(timestamp) для scores) — тогда удаление трассировок за месяц сведётся к одной операции DROP PARTITION.

   2. Затем разверните Advanced settings, выберите для Engine значение ReplacingMergeTree, чтобы обеспечить дедупликацию, и укажите updated_at в качестве столбца Version.

   

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

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

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

5. Настройте разрешения

   ClickPipes создаёт отдельного пользователя для записи данных в целевую таблицу. Выберите для этого внутреннего пользователя пользовательскую роль или одну из предопределённых ролей:

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

   

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

6. Завершите настройку

   Готово! Сначала ClickPipes выполнит дозагрузку исторических данных из всех файлов по указанному пути, а затем начнёт принимать новые файлы по мере их поступления в бакет.

   

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

​

Запросы к данным трассировки

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

​

Агрегация метрик по временным интервалам

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

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;

​

Восстановление трассировки

SELECT *
FROM observations_v2 FINAL
WHERE trace_id = '<trace_id>'
ORDER BY start_time;

​

Анализ качества

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;

​

См. также

• Langfuse и ClickHouse
• Экспорт данных Langfuse Cloud в Blob Storage
• ClickPipes для объектного хранилища