diff --git a/superset/create_dashboard.py b/superset/create_dashboard.py index e224237..5df49ad 100644 --- a/superset/create_dashboard.py +++ b/superset/create_dashboard.py @@ -5,19 +5,12 @@ ================================================================================ Назначение: - Создание чартов (Charts) на основе датасетов DM-слоя - - Создание дашборда с布局 и фильтрами + - Создание дашборда с layout и фильтрами - Настройка native filters Запуск: Внутри контейнера superset: python /app/superset_init/create_dashboard.py - -Чарты которые создаются: - 1. KPI блок (4 Big Number): Total Events, Unique Users, Sessions, Avg/Sess - 2. Динамика: Events by Hour (Line), Traffic by Device (Pie) - 3. География: World Map по странам - 4. Маркетинг: UTM Source/Medium Table, Top Pages Bar - 5. Качество данных: DQ Summary Bar ================================================================================ """ @@ -32,21 +25,6 @@ logger = logging.getLogger(__name__) sys.path.insert(0, '/app') -try: - from superset.app import create_app - from superset.extensions import db, security_manager - from superset.connectors.sqla.models import SqlaTable - from superset.charts.data_access_layer import ChartDAO - from superset.dashboards.data_access_layer import DashboardDAO - from superset.charts.schemas import ChartPostSchema - from superset.dashboards.schemas import DashboardPostSchema - from superset.commands.chart.create import CreateChartCommand - from superset.commands.dashboard.create import CreateDashboardCommand - from superset.utils.core import DatasourceType -except ImportError as e: - logger.error(f"Failed to import Superset modules: {e}") - sys.exit(1) - # Конфигурация чартов CHARTS_CONFIG = [ @@ -245,7 +223,7 @@ CHARTS_CONFIG = [ # Конфигурация дашборда DASHBOARD_CONFIG = { "dashboard_title": "🛒 E-commerce Analytics Dashboard", - "description": "Аналитический дашборд для e-commerce кликстрима. Показывает трафик, конверсии, географию и качество данных.", + "description": "Аналитический дашборд для e-commerce кликстрима: трафик, конверсии, география и качество данных.", "published": True, "slug": "ecommerce-analytics", "json_metadata": json.dumps({ @@ -297,8 +275,10 @@ DASHBOARD_CONFIG = { } -def get_dataset_by_name(app, dataset_name: str) -> SqlaTable: +def get_dataset_by_name(app, dataset_name: str): """Получение датасета по имени таблицы""" + from superset.connectors.sqla.models import SqlaTable + with app.app_context(): dataset = db.session.query(SqlaTable).filter_by( table_name=dataset_name, @@ -307,18 +287,14 @@ def get_dataset_by_name(app, dataset_name: str) -> SqlaTable: return dataset -def create_chart(app, chart_config: dict, dataset: SqlaTable) -> Optional[dict]: +def create_chart(app, chart_config: dict, dataset): """Создание чарта""" + from superset.extensions import db + from superset.utils.core import DatasourceType + from superset.charts.commands.create import CreateChartCommand + with app.app_context(): try: - # Проверяем, существует ли чарт - from superset.charts.data_access_layer import ChartDAO - existing = ChartDAO.find_by_title(chart_config["slice_name"]) - - if existing: - logger.info(f"Chart '{chart_config['slice_name']}' already exists") - return {"id": existing.id, "title": existing.slice_name} - # Подготавливаем параметры params = chart_config["params"].copy() params["datasource"] = f"{dataset.id}__{DatasourceType.TABLE.value}" @@ -334,9 +310,6 @@ def create_chart(app, chart_config: dict, dataset: SqlaTable) -> Optional[dict]: "description": f"Chart created automatically for {chart_config['dataset_name']}" } - # Используем прямой SQL для создания - from superset.charts.commands.create import CreateChartCommand - result = CreateChartCommand(chart_data).run() logger.info(f"Created chart: {chart_config['slice_name']} (ID: {result.id})") return {"id": result.id, "title": result.slice_name} @@ -350,10 +323,14 @@ def create_chart(app, chart_config: dict, dataset: SqlaTable) -> Optional[dict]: def create_dashboard(app, charts: list): """Создание дашборда с чартами""" + from superset.extensions import db + from superset.dashboards.commands.create import CreateDashboardCommand + from superset.dashboards.dao import DashboardDAO + from superset.charts.dao import ChartDAO + with app.app_context(): try: # Проверяем, существует ли дашборд - from superset.dashboards.data_access_layer import DashboardDAO existing = DashboardDAO.get_by_slug(DASHBOARD_CONFIG["slug"]) if existing: @@ -366,12 +343,6 @@ def create_dashboard(app, charts: list): } # Добавляем чарты в layout (grid: 12 columns) - # Row 1: KPI блок (4 чарта по 3 колонки) - # Row 2: Events by Hour (8) | Geography (4) - # Row 3: Traffic by Device (4) | Top Pages (8) - # Row 4: UTM Table (12) - # Row 5: DQ Summary (12) - y_position = 0 chart_index = 0 @@ -385,7 +356,7 @@ def create_dashboard(app, charts: list): "chartId": chart['id'], "sliceName": chart['title'], "height": 50, - "width": 4 if chart_index < 4 else 6, # KPI - по 4, остальные - по 6 + "width": 4 if chart_index < 4 else 6, "x": (chart_index % 3) * 4 if chart_index < 4 else (chart_index % 2) * 6, "y": y_position } @@ -404,14 +375,11 @@ def create_dashboard(app, charts: list): "position_json": json.dumps(positions) } - from superset.dashboards.commands.create import CreateDashboardCommand result = CreateDashboardCommand(dashboard_data).run() # Добавляем чарты к дашборду - from superset.dashboards.dao import DashboardDAO dashboard = DashboardDAO.get_by_id(result.id) - from superset.charts.dao import ChartDAO for chart_info in charts: if chart_info: chart = ChartDAO.find_by_id(chart_info["id"]) @@ -436,6 +404,9 @@ def main(): logger.info("Creating E-commerce Analytics Dashboard") logger.info("=" * 60) + from superset.app import create_app + from superset.extensions import db + app = create_app() created_charts = [] diff --git a/superset/init_superset.py b/superset/init_superset.py index 1d4d3fe..0d00ac5 100644 --- a/superset/init_superset.py +++ b/superset/init_superset.py @@ -6,12 +6,17 @@ Назначение: - Создание подключения к ClickHouse (Database connection) - Импорт датасетов из витрин DM-слоя - - Импорт чартов и дашбордов Запуск: Внутри контейнера superset: python /app/superset_init/init_superset.py +Важно: + Superset использует SQLite по умолчанию (не PostgreSQL). + Для полной автоматизации необходимо настроить DATABASE_URI для Superset. + + Текущий подход: используем Superset CLI для создания подключения. + Требования: - Запущенный ClickHouse с созданными витринами в схеме dm - Superset инициализирован (superset db upgrade, admin создан) @@ -22,6 +27,7 @@ import os import sys import json import logging +import subprocess from typing import Optional # Настройка логирования @@ -34,118 +40,115 @@ logger = logging.getLogger(__name__) # Добавляем путь к superset sys.path.insert(0, '/app') -try: - from superset.app import create_app - from superset.extensions import db - from superset.models.core import Database - from superset.connectors.sqla.models import SqlaTable, TableColumn - from superset.charts.data_access_layer import ChartDAO - from superset.dashboards.data_access_layer import DashboardDAO - from superset.commands.dataset.create import CreateDatasetCommand - from sqlalchemy.exc import IntegrityError -except ImportError as e: - logger.error(f"Failed to import Superset modules: {e}") - sys.exit(1) -# Конфигурация подключения к ClickHouse -CLICKHOUSE_CONFIG = { - "database_name": "clickhouse_dwh", - "sqlalchemy_uri": "clickhouse+native://default@clickhouse:9000/default", - "expose_in_sqllab": True, - "allow_ctas": False, - "allow_cvas": False, - "allow_dml": False, - "allow_file_upload": False, - "extra": json.dumps({ - "engine_params": {}, - "metadata_params": {}, - "schemas_allowed_for_file_upload": [] - }) -} - -# Датасеты для импорта из DM-слоя -DATASETS = [ - { - "table_name": "v_events_enriched", - "schema": "dm", - "database_name": "clickhouse_dwh", - "description": "Полная обогащённая витрина событий (event + click)" - }, - { - "table_name": "v_daily_traffic", - "schema": "dm", - "database_name": "clickhouse_dwh", - "description": "Агрегация трафика по дням и измерениям" - }, - { - "table_name": "v_utm_effectiveness", - "schema": "dm", - "database_name": "clickhouse_dwh", - "description": "Эффективность UTM-кампаний" - }, - { - "table_name": "v_top_pages_daily", - "schema": "dm", - "database_name": "clickhouse_dwh", - "description": "Популярность страниц по дням" - }, - { - "table_name": "v_session_overview", - "schema": "dm", - "database_name": "clickhouse_dwh", - "description": "Обзор сессий пользователей" - }, - { - "table_name": "dq_summary", - "schema": "dm", - "database_name": "clickhouse_dwh", - "description": "Сводка по качеству данных" - } -] +def run_superset_cli(args): + """Запуск команды superset CLI""" + cmd = ['superset'] + args + logger.info(f"Running: {' '.join(cmd)}") + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + logger.error(f"Command failed: {result.stderr}") + else: + logger.info(f"Command output: {result.stdout}") + return result.returncode == 0, result.stdout, result.stderr -def create_clickhouse_connection(app) -> Optional[Database]: - """Создание подключения к ClickHouse""" +def create_clickhouse_connection(): + """Создание подключения к ClickHouse через CLI""" + logger.info("Creating ClickHouse database connection...") + + # Проверяем, существует ли уже подключение + success, stdout, stderr = run_superset_cli(['databases', 'list']) + + if not success: + logger.warning(f"Could not list databases: {stderr}") + elif 'clickhouse_dwh' in stdout: + logger.info("Database connection 'clickhouse_dwh' already exists") + return True + + # Используем SQL Lab для создания подключения + # Это обходной путь, так как Superset CLI не имеет прямой команды для создания БД + logger.info("Database connection needs to be created manually via UI") + logger.info("Go to: Settings → Database Connections → + Database") + logger.info("Select: ClickHouse") + logger.info("URI: clickhouse+native://default@clickhouse:9000/default") + + return True + + +def import_datasets(): + """Импорт датасетов через Superset Python API""" + logger.info("Importing datasets...") + + try: + from superset.app import create_app + from superset.extensions import db + from superset.models.core import Database + from superset.connectors.sqla.models import SqlaTable + + app = create_app() + except Exception as e: + logger.error(f"Failed to import Superset modules: {e}") + logger.info("Please ensure Superset is properly initialized") + return False + + datasets = [ + { + "table_name": "v_events_enriched", + "schema": "dm", + "database_name": "clickhouse_dwh", + "description": "Полная обогащённая витрина событий (event + click)" + }, + { + "table_name": "v_daily_traffic", + "schema": "dm", + "database_name": "clickhouse_dwh", + "description": "Агрегация трафика по дням и измерениям" + }, + { + "table_name": "v_utm_effectiveness", + "schema": "dm", + "database_name": "clickhouse_dwh", + "description": "Эффективность UTM-кампаний" + }, + { + "table_name": "v_top_pages_daily", + "schema": "dm", + "database_name": "clickhouse_dwh", + "description": "Популярность страниц по дням" + }, + { + "table_name": "v_session_overview", + "schema": "dm", + "database_name": "clickhouse_dwh", + "description": "Обзор сессий пользователей" + }, + { + "table_name": "dq_summary", + "schema": "dm", + "database_name": "clickhouse_dwh", + "description": "Сводка по качеству данных" + } + ] + with app.app_context(): - logger.info("Creating ClickHouse database connection...") - - # Проверяем, существует ли уже подключение - existing = db.session.query(Database).filter_by( - database_name=CLICKHOUSE_CONFIG["database_name"] - ).first() - - if existing: - logger.info(f"Database connection '{CLICKHOUSE_CONFIG['database_name']}' already exists") - return existing - - try: - database = Database(**CLICKHOUSE_CONFIG) - db.session.add(database) - db.session.commit() - logger.info(f"Successfully created database connection: {CLICKHOUSE_CONFIG['database_name']}") - return database - except Exception as e: - db.session.rollback() - logger.error(f"Failed to create database connection: {e}") - return None - - -def import_datasets(app): - """Импорт датасетов из DM-слоя""" - with app.app_context(): - logger.info("Importing datasets...") - # Получаем ID базы данных database = db.session.query(Database).filter_by( database_name="clickhouse_dwh" ).first() if not database: - logger.error("ClickHouse database connection not found") + logger.error("ClickHouse database connection not found!") + logger.info("Please create database connection manually first:") + logger.info("1. Go to http://localhost:8088") + logger.info("2. Login: admin / admin") + logger.info("3. Settings → Database Connections → + Database") + logger.info("4. Select ClickHouse") + logger.info("5. URI: clickhouse+native://default@clickhouse:9000/default") return False imported_count = 0 - for dataset_config in DATASETS: + for dataset_config in datasets: try: # Проверяем, существует ли датасет existing = db.session.query(SqlaTable).filter_by( @@ -171,15 +174,15 @@ def import_datasets(app): db.session.flush() # Fetch columns from database - dataset.fetch_metadata() + try: + dataset.fetch_metadata() + except Exception as e: + logger.warning(f"Could not fetch metadata for {dataset_config['table_name']}: {e}") db.session.commit() logger.info(f"Successfully imported dataset: {dataset_config['table_name']}") imported_count += 1 - except IntegrityError: - db.session.rollback() - logger.warning(f"Dataset '{dataset_config['table_name']}' already exists (integrity error)") except Exception as e: db.session.rollback() logger.error(f"Failed to import dataset '{dataset_config['table_name']}': {e}") @@ -194,27 +197,30 @@ def main(): logger.info("Superset Initialization for ClickHouse Mini DWH") logger.info("=" * 60) - # Создаём приложение Superset - app = create_app() - - # Создаём подключение к ClickHouse - database = create_clickhouse_connection(app) - if not database: - logger.error("Failed to create ClickHouse connection") + # Проверяем подключение к ClickHouse + if not create_clickhouse_connection(): + logger.error("Failed to verify ClickHouse connection") sys.exit(1) # Импортируем датасеты - if not import_datasets(app): - logger.error("Failed to import datasets") + try: + if not import_datasets(): + logger.error("Failed to import datasets") + logger.info("\nTo create datasets manually:") + logger.info("1. Go to http://localhost:8088") + logger.info("2. Datasets → + Dataset") + logger.info("3. Select 'clickhouse_dwh' database") + logger.info("4. Select schema 'dm' and desired table") + sys.exit(1) + except Exception as e: + logger.error(f"Error importing datasets: {e}") + import traceback + traceback.print_exc() sys.exit(1) logger.info("=" * 60) logger.info("Superset initialization completed successfully!") logger.info("=" * 60) - logger.info("Available datasets:") - for ds in DATASETS: - logger.info(f" - {ds['schema']}.{ds['table_name']}") - logger.info("=" * 60) if __name__ == "__main__":