Kommo + Tableau: BI-дашборд продаж через ETL без ручного экспорта данных

Tableau — лидер enterprise BI-визуализации (Gartner Magic Quadrant Leader 2024): drag-and-drop аналитика, Tableau Server / Tableau Cloud, 80+ native connectors. Широко используется в enterprise где аналитика продаж, финансов и операций живёт в одном инструменте. Прямого Tableau connector для Kommo не существует — Tableau не имеет маркетплейса CRM-коннекторов на уровне native API. Правильная архитектура: выгрузка данных Kommo в реляционную базу через Python ETL, затем Tableau подключается к этой базе как к datasource. В отличие от Grafana или Metabase, Tableau — инструмент для аналитиков без SQL, с drag-and-drop построением визуализаций.

Tableau vs Grafana vs Metabase для CRM-аналитики

Tableau выбирают enterprise-компании где BI-аналитики строят дашборды для бизнес-пользователей без SQL, и где уже есть корпоративная лицензия.

Архитектура: Kommo -> Tableau

Kommo REST API  ->  Python ETL (cron)  ->  Postgres  ->  Tableau (Published Datasource)

Два варианта публикации данных в Tableau:

1. PostgreSQL datasource — Tableau подключается к Postgres напрямую. Просто, данные всегда свежие. Требует сетевого доступа Tableau Server -> Postgres.
2. Tableau Hyper API — Python публикует .hyper файл (in-memory columnar format) в Tableau Server/Cloud. Faster query, не требует постоянного DB-соединения. Оптимально для Tableau Cloud где Postgres недоступен напрямую.

ETL: Kommo -> Postgres

Базовый ETL аналогичен другим BI-интеграциям. Ключевые таблицы для Tableau дашборда продаж:

import requests
import psycopg2
from datetime import datetime, timezone, timedelta

KOMMO_BASE  = "https://youraccount.kommo.com/api/v4"
KOMMO_HDRS  = {"Authorization": "Bearer your_token"}
DB_DSN      = "postgresql://user:pass@localhost:5432/analytics"

# Схема: kommo_leads, kommo_pipelines, kommo_statuses, kommo_users
CREATE_TABLES = [
    (
        "kommo_leads",
        "id BIGINT PRIMARY KEY, name TEXT, status_id INT, pipeline_id INT, "
        "responsible_user_id BIGINT, price NUMERIC, "
        "created_at TIMESTAMPTZ, updated_at TIMESTAMPTZ, closed_at TIMESTAMPTZ, "
        "loss_reason_id INT, is_deleted BOOLEAN DEFAULT FALSE"
    ),
    (
        "kommo_statuses",
        "id INT PRIMARY KEY, pipeline_id INT, name TEXT, sort INT, type TEXT"
    ),
    (
        "kommo_users",
        "id BIGINT PRIMARY KEY, name TEXT, email TEXT, role TEXT"
    ),
]

def init_schema(conn):
    cur = conn.cursor()
    for table, cols in CREATE_TABLES:
        cur.execute(
            f"CREATE TABLE IF NOT EXISTS {table} ({cols})"
        )
    conn.commit()
    cur.close()

def fetch_leads_incremental(updated_after: datetime) -> list:
    ts, leads, page = int(updated_after.timestamp()), [], 1
    while True:
        resp = requests.get(f"{KOMMO_BASE}/leads",
                            headers=KOMMO_HDRS,
                            params={"updated_at[from]": ts, "page": page, "limit": 250})
        if resp.status_code == 204:
            break
        resp.raise_for_status()
        batch = resp.json().get("_embedded", {}).get("leads", [])
        if not batch:
            break
        leads.extend(batch)
        page += 1
    return leads

def upsert_leads(conn, leads: list):
    cur = conn.cursor()
    sql = (
        "INSERT INTO kommo_leads "
        "(id, name, status_id, pipeline_id, responsible_user_id, price, "
        "created_at, updated_at, closed_at, loss_reason_id) "
        "VALUES (%s,%s,%s,%s,%s,%s, "
        "to_timestamp(%s),to_timestamp(%s),to_timestamp(%s),%s) "
        "ON CONFLICT (id) DO UPDATE SET "
        "status_id=EXCLUDED.status_id, price=EXCLUDED.price, "
        "updated_at=EXCLUDED.updated_at, closed_at=EXCLUDED.closed_at, "
        "loss_reason_id=EXCLUDED.loss_reason_id, is_deleted=FALSE"
    )
    for lead in leads:
        cur.execute(sql, (
            lead["id"], lead.get("name"),
            lead.get("status_id"), lead.get("pipeline_id"),
            lead.get("responsible_user_id"), lead.get("price"),
            lead.get("created_at"), lead.get("updated_at"),
            lead.get("closed_at"), lead.get("loss_reason_id"),
        ))
    conn.commit()
    cur.close()

Публикация в Tableau Cloud через Hyper API

Если Tableau Cloud — Hyper API позволяет публиковать .hyper файл напрямую без постоянного DB-соединения:

# pip install tableauserverclient pantab
import tableauserverclient as TSC
import pantab

def publish_to_tableau_cloud(df, datasource_name: str):
    # df - pandas DataFrame с данными из Postgres
    server = TSC.Server("https://prod-apnortheast-a.online.tableau.com")
    tableau_auth = TSC.PersonalAccessTokenAuth(
        token_name  = "your_pat_name",
        token_value = "your_pat_value",
        site_id     = "your_site_id",
    )
    with server.auth.sign_in(tableau_auth):
        project_id = get_project_id(server, "Sales Analytics")
        hyper_path = f"/tmp/{datasource_name}.hyper"

        # pantab конвертирует DataFrame -> .hyper файл
        pantab.frame_to_hyper(df, hyper_path, table="kommo_leads")

        ds      = TSC.DatasourceItem(project_id, name=datasource_name)
        server.datasources.publish(
            ds, hyper_path,
            mode=TSC.Server.PublishMode.Overwrite,
        )
        print(f"Published: {datasource_name}")

def get_project_id(server, project_name: str) -> str:
    all_projects, _ = server.projects.get()
    for p in all_projects:
        if p.name == project_name:
            return p.id
    raise ValueError(f"Project '{project_name}' not found")

Ключевые метрики для Tableau дашборда продаж

После публикации datasource в Tableau аналитики строят дашборды drag-and-drop. Базовые метрики из kommo_leads:

• Win Rate — COUNTD(IF status_id = [won_id] THEN id END) / COUNTD(id)
• Avg Deal Size — AVG(IF status_id = [won_id] THEN price END)
• Sales Cycle Length — AVG(DATEDIFF('day', created_at, closed_at))
• Pipeline by Stage — bar chart по status_id, grouped by pipeline_id
• Revenue by Manager — join с kommo_users, SUM(price) per responsible_user_id
• Lead Velocity — COUNTD(id) per week с time filter

Calculated fields в Tableau создаются через UI — SQL не нужен аналитику.

Реальный кейс

Enterprise дистрибьютор (US, 300 человек, Tableau Server + Kommo):

• До: аналитик еженедельно экспортировал CSV из Kommo (ручной экспорт) -> загружал в Tableau. Данные 3–7 дней устаревшие. 2 часа работы в неделю.
• После: Python ETL раз в час -> Postgres -> Tableau PostgreSQL datasource (live connection). Данные свежее 1 часа. 0 ручной работы.
• Дополнительно: Tableau алерты (Data-driven Alerts) на снижение win rate ниже 25% -> email VP Sales автоматически.

Для кого актуально

• Enterprise с корпоративной лицензией Tableau где BI-команда строит дашборды для бизнеса
• Компании где аналитики продаж не пишут SQL, нужен drag-and-drop инструмент
• Организации где данные Kommo должны объединяться с другими источниками (ERP, финансы) в одном Tableau дашборде
• Teams где уже есть Tableau Server/Cloud и добавление CRM — расширение существующей экосистемы

Часто задаваемые вопросы

Tableau live connection vs extract — что выбрать для Kommo данных?

Live connection: Tableau запрашивает Postgres каждый раз при открытии дашборда. Актуальные данные, медленнее при большом объёме. Extract: Tableau делает снимок данных по расписанию (hourly/daily), хранит в .tde/.hyper формате. Быстрее, данные с задержкой. Для Kommo (обычно до 100k лидов) — live connection + materialized views в Postgres для производительности.

Tableau Prep Builder vs Python ETL — что лучше?

Tableau Prep Builder — визуальный ETL инструмент Tableau (drag-and-drop трансформации). Не поддерживает Kommo API нативно — только SQL-источники и файлы. Для Kommo данных: Python ETL -> Postgres (обязательный шаг), далее Tableau Prep Builder может дополнительно трансформировать данные если нужны сложные joins. Для простых дашборд продаж Python ETL -> Postgres достаточно без Prep.

Как настроить Tableau Server (не Cloud) с Postgres?

Tableau Server -> Data Sources -> Connect to Data -> PostgreSQL -> указать host/port/db/user/password. Если Postgres на том же сервере — localhost. Если на отдельном — нужен сетевой доступ (TCP 5432). ODBC PostgreSQL драйвер устанавливается на Tableau Server автоматически при установке.

Можно ли обновлять Tableau Extract автоматически после каждого ETL?

Да. Tableau REST API: POST /api/{version}/sites/{siteId}/datasources/{id}/refreshes инициирует refresh extract. Добавить в конец Python ETL-скрипта:

# trigger_tableau_refresh(datasource_id)
resp = requests.post(
    f"{tableau_server}/api/3.15/sites/{site_id}/datasources/{ds_id}/refreshes",
    headers={"x-tableau-auth": token},
    json={},
)

Итого

• Архитектура: Kommo API -> Python ETL (incremental, cron) -> Postgres -> Tableau
• Два пути в Tableau: PostgreSQL live connection (проще) или Hyper API (Tableau Cloud)
• Tableau Cloud -> pantab + tableauserverclient для публикации .hyper файлов
• Ключевые таблицы: kommo_leads, kommo_statuses, kommo_users, kommo_pipelines
• Tableau алерты (Data-driven) — автоматические уведомления при отклонении метрик

Если вы используете Tableau и хотите подключить данные Kommo без еженедельного CSV-экспорта — опишите объём данных и используете Tableau Server или Cloud. Exceltic.dev настроит ETL и published datasource.