Daft 多模态数据处理平台

功能概览

项目结构

daft-platform/ # 项目根目录
├── README.md
├── requirements.txt
├── Makefile # make dev / make test / make run
├── .gitignore
└── daft_platform/ # Python 包
 ├── app.py # FastAPI 应用入口
 ├── config.py # 配置管理(环境变量)
 ├── models/ # Pydantic 请求/响应模型
 │ ├── sql.py # SQL 执行相关
 │ ├── datasource.py # 数据源/元数据相关
 │ └── function.py # 函数注册表相关
 ├── routers/ # API 路由层
 │ ├── sql.py # /api/sql/*
 │ ├── datasource.py # /api/datasource/*
 │ ├── function.py # /api/function/*
 │ └── settings.py # /api/settings/*
 ├── services/ # 业务逻辑层
 │ ├── session_manager.py # Daft Session 单例管理
 │ ├── sql_executor.py # SQL 执行 + 错误翻译
 │ ├── sql_history.py # 执行历史(SQLite)
 │ ├── function_registry.py # 函数注册表
 │ ├── builtin_udfs.py # 16 个预置 AI UDF
 │ ├── metadata_service.py # 元数据抽象接口 + 工厂函数
 │ ├── gravitino_metadata.py # Gravitino REST API 实现
 │ └── local_metadata.py # SQLite 本地降级实现
 ├── templates/ # Jinja2 前端页面
 │ ├── index.html # SQL 工作台主页
 │ └── dag.html # DAG 编排页
 ├── static/ # 静态资源(JS/CSS)
 ├── tests/ # 单元测试(pytest)
 ├── test_data/ # 测试用图片(dog_*.jpg)
 ├── test_data.csv # 测试用 CSV
 ├── deploy/ # 部署配置
 │ └── gravitino-pod.yaml # Gravitino K8s 部署
 └── data/ # 运行时数据(.gitignore)

技术栈

快速开始

1. 前置条件

# 编译安装 Daft 引擎
git clone https://github.com/Eventual-Inc/Daft.git && cd Daft
make .venv && make build
source .venv/bin/activate

2. 克隆本项目并安装依赖

git clone https://github.com/fightBoxing/daft-platform.git
cd daft-platform
pip install -r requirements.txt
# 可选:AI UDF 额外依赖
pip install easyocr # OCR 文字识别
pip install imagehash # 图像感知哈希
pip install face_recognition # 人脸检测(或 opencv-python)

3. 启动服务

# 开发模式(自动重载)
make dev
# 生产模式
make run
# 或直接用 uvicorn
uvicorn daft_platform.app:app --host 0.0.0.0 --port 8080

配置

AI / LLM 配置

元数据服务降级

export DAFT_PLATFORM_GRAVITINO_ENABLED=false

部署 Gravitino(可选)

# K8s 部署
kubectl apply -f daft_platform/deploy/gravitino-pod.yaml
# Docker 部署
docker run -d --name gravitino -p 30090:8090 apache/gravitino:latest

API 接口

SQL 执行

# 执行 SQL
curl -X POST http://localhost:8080/api/sql/execute \
 -H 'Content-Type: application/json' \
 -d '{"sql": "SELECT * FROM my_table LIMIT 10", "limit": 1000}'
# SQL 安全验证
curl -X POST http://localhost:8080/api/sql/validate \
 -H 'Content-Type: application/json' \
 -d '{"sql": "SELECT 1"}'
# 查看执行历史
curl http://localhost:8080/api/sql/history?limit=10

数据源管理

# 注册 JSON 数据表
curl -X POST http://localhost:8080/api/datasource/register \
 -H 'Content-Type: application/json' \
 -d '{
 "name": "users",
 "source_type": "json",
 "data": {"id": [1, 2, 3], "name": ["Alice", "Bob", "Carol"]}
 }'
# 注册文件表(CSV/Parquet/JSON)
curl -X POST http://localhost:8080/api/datasource/register \
 -H 'Content-Type: application/json' \
 -d '{"name": "sales", "source_type": "file", "path": "/data/sales.parquet"}'
# 注册 HuggingFace 数据集
curl -X POST http://localhost:8080/api/datasource/register \
 -H 'Content-Type: application/json' \
 -d '{"name": "mnist", "source_type": "huggingface", "dataset": "mnist", "limit": 1000}'
# 注册 Glob 路径(批量文件)
curl -X POST http://localhost:8080/api/datasource/register \
 -H 'Content-Type: application/json' \
 -d '{"name": "images", "source_type": "glob", "glob_pattern": "/data/photos/**/*.jpg"}'
# 查看元数据树
curl http://localhost:8080/api/datasource/tree

函数查询

# 列出所有函数
curl http://localhost:8080/api/function/list
# 按分类筛选
curl http://localhost:8080/api/function/list?category=ai
# 搜索函数
curl http://localhost:8080/api/function/search?q=image
# SQL 自动补全
curl http://localhost:8080/api/function/completions?prefix=image_

SQL 语法指南

基础查询

-- 注册测试数据后查询
SELECT * FROM test_data WHERE score > 85 ORDER BY score DESC;
-- 聚合统计
SELECT category, COUNT(*) AS cnt, AVG(score) AS avg_score
FROM test_data
GROUP BY category
ORDER BY avg_score DESC;
-- CTE(公用表表达式)
WITH top_students AS (
 SELECT * FROM test_data WHERE score > 90
)
SELECT category, COUNT(*) FROM top_students GROUP BY category;
-- 窗口函数
SELECT name, score,
 RANK() OVER (ORDER BY score DESC) AS rank,
 ROW_NUMBER() OVER (PARTITION BY category ORDER BY score DESC) AS cat_rank
FROM test_data;

字符串函数

SELECT lower(name), upper(city) FROM test_data;
SELECT * FROM test_data WHERE contains(city, '京');
SELECT * FROM test_data WHERE starts_with(name, '王');
SELECT length(name) AS name_len FROM test_data;
SELECT replace(city, '北京', 'Beijing') FROM test_data;
SELECT split(name, '') FROM test_data; -- 按字符拆分

数值函数

SELECT abs(score - 85) AS diff FROM test_data;
SELECT round(score, 0) AS rounded FROM test_data;
SELECT ceil(score), floor(score) FROM test_data;
SELECT clip(score, 80, 95) AS clipped FROM test_data;

时间函数

SELECT current_date(), current_timestamp();
SELECT year(to_date('2024年03月15日', '%Y-%m-%d'));
SELECT date_trunc('month', current_timestamp());

列表函数

SELECT explode(split(name, '')) FROM test_data; -- 展开为多行
SELECT list_count(split(city, '')) FROM test_data;

多模态数据处理 SQL

图像处理

-- 1. 注册图片数据集(Glob 方式扫描目录)
-- 在 Web 界面中:数据源 → 注册 → Glob → 路径填 /path/to/images/**/*.jpg
-- 2. 解码图片并调整尺寸
SELECT path, image_decode(path) AS img FROM images;
SELECT path, image_resize(image_decode(path), 224, 224) AS resized FROM images;
-- 3. 图像裁剪
SELECT path, image_crop(image_decode(path), 0, 0, 100, 100) AS cropped FROM images;

AI 视觉识别(需配置 LLM API Key)

-- 图像内容识别:调用 Vision API 返回中文描述
SELECT path, classify_image(path) AS description FROM images;
-- 图像元信息:宽高、格式、文件大小
SELECT path, image_info(path) AS info FROM images;
-- 主色调提取:返回 HEX 颜色值
SELECT path, dominant_color(path) AS color FROM images;
-- 图像感知哈希:用于去重/相似度比较
SELECT path, image_phash(path) AS phash FROM images;
-- 人脸检测:返回人脸数量
SELECT path, face_count(path) AS faces FROM images;
-- EXIF 元数据:相机型号、GPS、拍摄时间等
SELECT path, exif_extract(path) AS exif FROM images;
-- OCR 文字识别(支持中英文)
SELECT path, ocr_extract(path) AS text FROM document_images;

完整图像分析流水线示例

-- 假设已注册 images 表(Glob: /data/photos/**/*.jpg)
-- 分析每张图片:识别内容 + 提取信息 + 主色调 + 人脸
SELECT
 path,
 classify_image(path) AS description,
 image_info(path) AS info,
 dominant_color(path) AS color,
 face_count(path) AS faces,
 image_phash(path) AS phash
FROM images
LIMIT 20;
-- 筛选包含人脸的图片
SELECT path, face_count(path) AS faces
FROM images
WHERE face_count(path) > 0;
-- 图像去重(基于感知哈希)
SELECT phash, COUNT(*) AS cnt, MIN(path) AS sample
FROM (SELECT path, image_phash(path) AS phash FROM images)
GROUP BY phash
HAVING cnt > 1;

文本分析

-- 情感分析:返回 positive / negative / neutral
SELECT name, sentiment(name) AS mood FROM test_data;
-- 语言检测:返回 zh / en
SELECT name, detect_language(name) AS lang FROM test_data;
-- 关键词提取(TF 频率统计)
SELECT extract_keywords(description) AS keywords FROM articles;
-- 文本摘要(调用 LLM,50 字以内)
SELECT text_summary(content) AS summary FROM articles;
-- PII 脱敏(自动识别手机号/身份证/邮箱/银行卡)
SELECT mask_pii('联系方式: 13812345678, 邮箱: test@example.com') AS masked;
-- 输出: "联系方式: 138****5678, 邮箱: te***@example.com"
-- 文本相似度(Jaccard bigram)
SELECT text_similarity('你好世界', '你好中国') AS similarity;

音频 & 文件处理

-- 音频时长(秒,支持 mp3/wav/flac)
SELECT path, audio_duration(path) AS duration_sec FROM audio_files;
-- 文件 MD5 哈希
SELECT path, file_md5(path) AS md5 FROM files;
-- 文件转 Base64
SELECT path, to_base64(path) AS b64 FROM files;

向量距离计算

-- 余弦相似度(用于 Embedding 比较)
SELECT cosine_similarity(emb1, emb2) AS sim FROM embedding_pairs;
-- 欧氏距离
SELECT euclidean_distance(emb1, emb2) AS dist FROM embedding_pairs;
-- 点积
SELECT dot_product(emb1, emb2) AS dp FROM embedding_pairs;

URL & 网络

-- 从 URL 下载内容
SELECT url_download(url_col) AS content FROM web_sources;
-- 解析 URL 各部分
SELECT url_parse(url_col) AS parsed FROM web_sources;
-- 猜测文件 MIME 类型
SELECT guess_mime_type(path) AS mime FROM files;

内置 AI UDF 完整列表

测试

# 运行全部测试
make test
# 运行单个测试文件
python -m pytest daft_platform/tests/test_sql_executor.py -v
# 运行单个测试方法
python -m pytest daft_platform/tests/test_routers.py::test_register_and_list_table -v

安全机制

SQL 安全验证

DROP TABLE → ❌ 拦截
DELETE FROM → ❌ 拦截
TRUNCATE → ❌ 拦截
ALTER TABLE → ❌ 拦截
SELECT / WITH → ✅ 允许

错误翻译

已知问题

License