feat(superset): автоматическая инициализация с PostgreSQL метаданными
- Добавлена автоматическая инициализация Superset (подключение ClickHouse, 6 датасетов, 10 чартов, дашборд) - Переведено хранение метаданных с SQLite на PostgreSQL (shared с Airflow) - Добавлен superset_config.py для конфигурации PostgreSQL - Обновлен Dockerfile.superset: postgresql-client, psycopg2-binary - Обновлен docker-compose.yml: volume mount конфига, SUPERSET_CONFIG_PATH - Исправлены скрипты init_superset.py и create_dashboard.py для работы с shell - Обновлена документация в README.md: раздел Superset с инструкциями Тестирование: - Проверена работа после перезапуска (данные сохраняются) - Проверен чистый запуск с нуля - API и UI доступны
This commit is contained in:
+137
-151
@@ -5,8 +5,7 @@
|
||||
================================================================================
|
||||
Назначение:
|
||||
- Создание чартов (Charts) на основе датасетов DM-слоя
|
||||
- Создание дашборда с layout и фильтрами
|
||||
- Настройка native filters
|
||||
- Создание дашборда с layout
|
||||
|
||||
Запуск:
|
||||
Внутри контейнера superset:
|
||||
@@ -275,167 +274,154 @@ DASHBOARD_CONFIG = {
|
||||
}
|
||||
|
||||
|
||||
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,
|
||||
schema="dm"
|
||||
).first()
|
||||
return dataset
|
||||
|
||||
|
||||
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:
|
||||
# Подготавливаем параметры
|
||||
params = chart_config["params"].copy()
|
||||
params["datasource"] = f"{dataset.id}__{DatasourceType.TABLE.value}"
|
||||
params["viz_type"] = chart_config["viz_type"]
|
||||
|
||||
# Создаём чарт через команду
|
||||
chart_data = {
|
||||
"slice_name": chart_config["slice_name"],
|
||||
"viz_type": chart_config["viz_type"],
|
||||
"datasource_id": dataset.id,
|
||||
"datasource_type": DatasourceType.TABLE.value,
|
||||
"params": json.dumps(params),
|
||||
"description": f"Chart created automatically for {chart_config['dataset_name']}"
|
||||
}
|
||||
|
||||
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}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to create chart '{chart_config['slice_name']}': {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return None
|
||||
|
||||
|
||||
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:
|
||||
# Проверяем, существует ли дашборд
|
||||
existing = DashboardDAO.get_by_slug(DASHBOARD_CONFIG["slug"])
|
||||
|
||||
if existing:
|
||||
logger.info(f"Dashboard '{DASHBOARD_CONFIG['dashboard_title']}' already exists")
|
||||
return existing
|
||||
|
||||
# Создаём позиции чартов для layout
|
||||
positions = {
|
||||
"DASHBOARD_VERSION_KEY": "v2"
|
||||
}
|
||||
|
||||
# Добавляем чарты в layout (grid: 12 columns)
|
||||
y_position = 0
|
||||
chart_index = 0
|
||||
|
||||
for chart in charts:
|
||||
if chart:
|
||||
positions[f"CHART-{chart['id']}"] = {
|
||||
"id": f"CHART-{chart['id']}",
|
||||
"type": "CHART",
|
||||
"parents": ["ROOT_ID"],
|
||||
"meta": {
|
||||
"chartId": chart['id'],
|
||||
"sliceName": chart['title'],
|
||||
"height": 50,
|
||||
"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
|
||||
}
|
||||
}
|
||||
chart_index += 1
|
||||
if chart_index % 4 == 0:
|
||||
y_position += 50
|
||||
|
||||
# Создаём дашборд
|
||||
dashboard_data = {
|
||||
"dashboard_title": DASHBOARD_CONFIG["dashboard_title"],
|
||||
"slug": DASHBOARD_CONFIG["slug"],
|
||||
"description": DASHBOARD_CONFIG["description"],
|
||||
"published": DASHBOARD_CONFIG["published"],
|
||||
"json_metadata": DASHBOARD_CONFIG["json_metadata"],
|
||||
"position_json": json.dumps(positions)
|
||||
}
|
||||
|
||||
result = CreateDashboardCommand(dashboard_data).run()
|
||||
|
||||
# Добавляем чарты к дашборду
|
||||
dashboard = DashboardDAO.get_by_id(result.id)
|
||||
|
||||
for chart_info in charts:
|
||||
if chart_info:
|
||||
chart = ChartDAO.find_by_id(chart_info["id"])
|
||||
if chart:
|
||||
dashboard.slices.append(chart)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
logger.info(f"Created dashboard: {DASHBOARD_CONFIG['dashboard_title']} (ID: {result.id})")
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to create dashboard: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
"""Главная функция"""
|
||||
logger.info("=" * 60)
|
||||
logger.info("Creating E-commerce Analytics Dashboard")
|
||||
logger.info("=" * 60)
|
||||
|
||||
# Импорты внутри main после создания app context
|
||||
from superset.app import create_app
|
||||
from superset.extensions import db
|
||||
|
||||
app = create_app()
|
||||
|
||||
created_charts = []
|
||||
|
||||
# Создаём чарты
|
||||
for chart_config in CHARTS_CONFIG:
|
||||
dataset = get_dataset_by_name(app, chart_config["dataset_name"])
|
||||
if not dataset:
|
||||
logger.warning(f"Dataset '{chart_config['dataset_name']}' not found, skipping chart")
|
||||
continue
|
||||
with app.app_context():
|
||||
from superset.extensions import db
|
||||
from superset.models.slice import Slice
|
||||
from superset.models.dashboard import Dashboard
|
||||
from superset.connectors.sqla.models import SqlaTable
|
||||
|
||||
chart = create_chart(app, chart_config, dataset)
|
||||
if chart:
|
||||
created_charts.append(chart)
|
||||
|
||||
logger.info(f"Created {len(created_charts)} charts")
|
||||
|
||||
# Создаём дашборд
|
||||
if created_charts:
|
||||
dashboard = create_dashboard(app, created_charts)
|
||||
if dashboard:
|
||||
logger.info("=" * 60)
|
||||
logger.info("Dashboard created successfully!")
|
||||
logger.info(f"Dashboard URL: /superset/dashboard/{dashboard.id}/")
|
||||
logger.info("=" * 60)
|
||||
created_charts = []
|
||||
|
||||
# Создаём чарты
|
||||
for chart_config in CHARTS_CONFIG:
|
||||
dataset = db.session.query(SqlaTable).filter_by(
|
||||
table_name=chart_config["dataset_name"],
|
||||
schema="dm"
|
||||
).first()
|
||||
|
||||
if not dataset:
|
||||
logger.warning(f"Dataset '{chart_config['dataset_name']}' not found, skipping chart")
|
||||
continue
|
||||
|
||||
try:
|
||||
# Проверяем, существует ли уже чарт
|
||||
existing = db.session.query(Slice).filter_by(
|
||||
slice_name=chart_config["slice_name"]
|
||||
).first()
|
||||
|
||||
if existing:
|
||||
logger.info(f"Chart '{chart_config['slice_name']}' already exists (ID: {existing.id})")
|
||||
created_charts.append({"id": existing.id, "title": existing.slice_name})
|
||||
continue
|
||||
|
||||
# Подготавливаем параметры
|
||||
params = chart_config["params"].copy()
|
||||
params["datasource"] = f"{dataset.id}__table"
|
||||
params["viz_type"] = chart_config["viz_type"]
|
||||
|
||||
# Создаём чарт
|
||||
chart = Slice(
|
||||
slice_name=chart_config["slice_name"],
|
||||
viz_type=chart_config["viz_type"],
|
||||
datasource_id=dataset.id,
|
||||
datasource_type="table",
|
||||
datasource_name=dataset.table_name,
|
||||
params=json.dumps(params),
|
||||
description=f"Chart created automatically for {chart_config['dataset_name']}"
|
||||
)
|
||||
|
||||
db.session.add(chart)
|
||||
db.session.flush()
|
||||
|
||||
logger.info(f"Created chart: {chart_config['slice_name']} (ID: {chart.id})")
|
||||
created_charts.append({"id": chart.id, "title": chart.slice_name})
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to create chart '{chart_config['slice_name']}': {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
db.session.rollback()
|
||||
|
||||
logger.info(f"Created/Found {len(created_charts)} charts")
|
||||
|
||||
# Создаём дашборд
|
||||
if created_charts:
|
||||
try:
|
||||
# Проверяем, существует ли дашборд
|
||||
existing = db.session.query(Dashboard).filter_by(
|
||||
slug=DASHBOARD_CONFIG["slug"]
|
||||
).first()
|
||||
|
||||
if existing:
|
||||
logger.info(f"Dashboard '{DASHBOARD_CONFIG['dashboard_title']}' already exists (ID: {existing.id})")
|
||||
logger.info("=" * 60)
|
||||
logger.info("Dashboard already exists!")
|
||||
logger.info(f"Dashboard URL: /superset/dashboard/{existing.id}/")
|
||||
logger.info("=" * 60)
|
||||
return
|
||||
|
||||
# Создаём позиции чартов для layout
|
||||
positions = {"DASHBOARD_VERSION_KEY": "v2"}
|
||||
|
||||
# Добавляем чарты в layout (grid: 12 columns)
|
||||
y_position = 0
|
||||
chart_index = 0
|
||||
|
||||
for chart in created_charts:
|
||||
if chart:
|
||||
positions[f"CHART-{chart['id']}"] = {
|
||||
"id": f"CHART-{chart['id']}",
|
||||
"type": "CHART",
|
||||
"parents": ["ROOT_ID"],
|
||||
"meta": {
|
||||
"chartId": chart['id'],
|
||||
"sliceName": chart['title'],
|
||||
"height": 50,
|
||||
"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
|
||||
}
|
||||
}
|
||||
chart_index += 1
|
||||
if chart_index % 4 == 0:
|
||||
y_position += 50
|
||||
|
||||
# Создаём дашборд
|
||||
dashboard = Dashboard(
|
||||
dashboard_title=DASHBOARD_CONFIG["dashboard_title"],
|
||||
slug=DASHBOARD_CONFIG["slug"],
|
||||
description=DASHBOARD_CONFIG["description"],
|
||||
published=DASHBOARD_CONFIG["published"],
|
||||
json_metadata=DASHBOARD_CONFIG["json_metadata"],
|
||||
position_json=json.dumps(positions)
|
||||
)
|
||||
|
||||
db.session.add(dashboard)
|
||||
db.session.flush()
|
||||
|
||||
# Добавляем чарты к дашборду
|
||||
for chart_info in created_charts:
|
||||
if chart_info:
|
||||
chart = db.session.query(Slice).filter_by(id=chart_info["id"]).first()
|
||||
if chart:
|
||||
dashboard.slices.append(chart)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
logger.info(f"Created dashboard: {DASHBOARD_CONFIG['dashboard_title']} (ID: {dashboard.id})")
|
||||
logger.info("=" * 60)
|
||||
logger.info("Dashboard created successfully!")
|
||||
logger.info(f"Dashboard URL: /superset/dashboard/{dashboard.id}/")
|
||||
logger.info("=" * 60)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to create dashboard: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
db.session.rollback()
|
||||
else:
|
||||
logger.error("Failed to create dashboard")
|
||||
else:
|
||||
logger.error("No charts created, cannot create dashboard")
|
||||
logger.error("No charts created, cannot create dashboard")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
+88
-90
@@ -13,9 +13,10 @@
|
||||
|
||||
Важно:
|
||||
Superset использует SQLite по умолчанию (не PostgreSQL).
|
||||
Для полной автоматизации необходимо настроить DATABASE_URI для Superset.
|
||||
|
||||
Текущий подход: используем Superset CLI для создания подключения.
|
||||
Текущий подход:
|
||||
1. CLI для создания подключения к БД
|
||||
2. Superset shell для импорта датасетов (требуется app context)
|
||||
|
||||
Требования:
|
||||
- Запущенный ClickHouse с созданными витринами в схеме dm
|
||||
@@ -57,41 +58,30 @@ def create_clickhouse_connection():
|
||||
"""Создание подключения к ClickHouse через CLI"""
|
||||
logger.info("Creating ClickHouse database connection...")
|
||||
|
||||
# Проверяем, существует ли уже подключение
|
||||
success, stdout, stderr = run_superset_cli(['databases', 'list'])
|
||||
# Создаем подключение через set-database-uri
|
||||
# clickhouse-connect использует HTTP порт 8123 внутри Docker сети
|
||||
success, stdout, stderr = run_superset_cli([
|
||||
'set-database-uri',
|
||||
'-d', 'clickhouse_dwh',
|
||||
'-u', 'clickhouse+connect://default@clickhouse:8123/default'
|
||||
])
|
||||
|
||||
if not success:
|
||||
logger.warning(f"Could not list databases: {stderr}")
|
||||
elif 'clickhouse_dwh' in stdout:
|
||||
logger.info("Database connection 'clickhouse_dwh' already exists")
|
||||
if success:
|
||||
logger.info("Successfully created ClickHouse database connection 'clickhouse_dwh'")
|
||||
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
|
||||
else:
|
||||
logger.error(f"Failed to create database connection: {stderr}")
|
||||
logger.info("Please create manually via UI:")
|
||||
logger.info("1. Go to: Settings → Database Connections → + Database")
|
||||
logger.info("2. Select: ClickHouse")
|
||||
logger.info("3. URI: clickhouse+connect://default@clickhouse:8123/default")
|
||||
return False
|
||||
|
||||
|
||||
def import_datasets():
|
||||
"""Импорт датасетов через Superset Python API"""
|
||||
"""Импорт датасетов через Superset shell"""
|
||||
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",
|
||||
@@ -131,64 +121,72 @@ def import_datasets():
|
||||
}
|
||||
]
|
||||
|
||||
with app.app_context():
|
||||
# Получаем ID базы данных
|
||||
database = db.session.query(Database).filter_by(
|
||||
database_name="clickhouse_dwh"
|
||||
).first()
|
||||
|
||||
if not database:
|
||||
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:
|
||||
try:
|
||||
# Проверяем, существует ли датасет
|
||||
existing = db.session.query(SqlaTable).filter_by(
|
||||
table_name=dataset_config["table_name"],
|
||||
schema=dataset_config["schema"]
|
||||
).first()
|
||||
|
||||
if existing:
|
||||
logger.info(f"Dataset '{dataset_config['table_name']}' already exists")
|
||||
continue
|
||||
|
||||
# Создаём датасет
|
||||
dataset = SqlaTable(
|
||||
table_name=dataset_config["table_name"],
|
||||
schema=dataset_config["schema"],
|
||||
database_id=database.id,
|
||||
database=database,
|
||||
description=dataset_config["description"],
|
||||
is_sqllab_view=False
|
||||
)
|
||||
|
||||
db.session.add(dataset)
|
||||
db.session.flush()
|
||||
|
||||
# Fetch columns from database
|
||||
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 Exception as e:
|
||||
db.session.rollback()
|
||||
logger.error(f"Failed to import dataset '{dataset_config['table_name']}': {e}")
|
||||
|
||||
logger.info(f"Imported {imported_count} new datasets")
|
||||
return True
|
||||
# Создаем Python скрипт для выполнения внутри superset shell
|
||||
script_lines = [
|
||||
"from superset.extensions import db",
|
||||
"from superset.models.core import Database",
|
||||
"from superset.connectors.sqla.models import SqlaTable",
|
||||
"",
|
||||
"# Получаем базу данных",
|
||||
"database = db.session.query(Database).filter_by(database_name='clickhouse_dwh').first()",
|
||||
"if not database:",
|
||||
" print('ERROR: Database clickhouse_dwh not found')",
|
||||
" exit(1)",
|
||||
"",
|
||||
"print(f'Found database: {database.database_name} (id={database.id})')",
|
||||
"",
|
||||
"imported = 0",
|
||||
]
|
||||
|
||||
for ds in datasets:
|
||||
script_lines.extend([
|
||||
"",
|
||||
f"# Dataset: {ds['table_name']}",
|
||||
f"existing = db.session.query(SqlaTable).filter_by(table_name='{ds['table_name']}', schema='{ds['schema']}').first()",
|
||||
"if existing:",
|
||||
f" print(f'Dataset {ds['table_name']} already exists')",
|
||||
"else:",
|
||||
" try:",
|
||||
f" dataset = SqlaTable(table_name='{ds['table_name']}', schema='{ds['schema']}', database_id=database.id, database=database, description='{ds['description']}')",
|
||||
" db.session.add(dataset)",
|
||||
" db.session.flush()",
|
||||
f" print(f'Created dataset: {ds['table_name']}')",
|
||||
" imported += 1",
|
||||
" except Exception as e:",
|
||||
f" print(f'Error creating {ds['table_name']}: {{e}}')",
|
||||
" db.session.rollback()",
|
||||
])
|
||||
|
||||
script_lines.extend([
|
||||
"",
|
||||
"db.session.commit()",
|
||||
"print(f'Successfully imported {imported} datasets')",
|
||||
])
|
||||
|
||||
script_content = '\n'.join(script_lines)
|
||||
|
||||
# Запускаем через superset shell
|
||||
cmd = ['superset', 'shell']
|
||||
logger.info("Running datasets import via superset shell...")
|
||||
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
input=script_content,
|
||||
capture_output=True,
|
||||
text=True
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
logger.error(f"Shell command failed: {result.stderr}")
|
||||
return False
|
||||
|
||||
logger.info(f"Shell output:\n{result.stdout}")
|
||||
if "ERROR" in result.stdout:
|
||||
logger.error("Failed to import some datasets")
|
||||
return False
|
||||
|
||||
logger.info("Datasets imported successfully")
|
||||
return True
|
||||
|
||||
|
||||
def main():
|
||||
@@ -197,9 +195,9 @@ def main():
|
||||
logger.info("Superset Initialization for ClickHouse Mini DWH")
|
||||
logger.info("=" * 60)
|
||||
|
||||
# Проверяем подключение к ClickHouse
|
||||
# Создаем подключение к ClickHouse
|
||||
if not create_clickhouse_connection():
|
||||
logger.error("Failed to verify ClickHouse connection")
|
||||
logger.error("Failed to create ClickHouse connection")
|
||||
sys.exit(1)
|
||||
|
||||
# Импортируем датасеты
|
||||
|
||||
Reference in New Issue
Block a user