🚀 Parquet MCP 服务器
Parquet MCP 服务器是一个用于处理 Parquet 文件的工具集，它提供了嵌入、信息获取、格式转换等功能，能帮助用户高效地处理和分析 Parquet 数据。
🚀 快速开始
项目结构
parquet-mcp-server/
├── README.md # 项目说明文件
├── src/ # 源代码目录
│ ├── __init__.py # 包初始化文件
│ ├── client.py # MCP客户端接口
│ └── tools/ # 工具模块
│ ├── embedding.py # 嵌入工具
│ ├── parquet_info.py # Parquet 文件信息工具
│ ├── duckdb_conversion.py # DuckDB 转换工具
│ ├── postgres_conversion.py # PostgreSQL 转换工具
│ └── markdown_processing.py # Markdown 处理工具
├── tests/ # 测试用例目录
│ ├── __init__.py # 测试包初始化文件
│ ├── test_embedding.py # 嵌入功能测试
│ ├── test_parquet_info.py # Parquet 文件信息工具测试
│ ├── test_duckdb_conversion.py # DuckDB 转换测试
│ ├── test_postgres_conversion.py # PostgreSQL 转换测试
│ └── test_markdown_processing.py # Markdown 处理测试
└── requirements.txt # 项目依赖管理文件
环境变量
在 .env 文件中添加以下环境变量：
# 必要配置
OPENAI_API_KEY=your_openai_api_key # OpenAI API 密钥
# 可选配置（根据需要调整）
OLLAMA_SERVER_URL=http://localhost:11434 # Ollama 服务器地址，默认为本地
安装与运行
使用命令行工具安装
pip install -r requirements.txt # 安装依赖
python src/run_server.py # 启动 MCP 服务器
使用测试客户端
from parquet_mcp_server.client import (
convert_to_duckdb,
embed_parquet,
get_parquet_info,
convert_to_postgres,
process_markdown_file
)
# 测试 DuckDB 转换
result = convert_to_duckdb(
parquet_path="input.parquet",
output_dir="db_output"
)
# 测试嵌入功能
result = embed_parquet(
input_path="input.parquet",
output_path="output.parquet",
column_name="text",
embedding_column="embeddings",
batch_size=2
)
# 测试 Parquet 信息工具
result = get_parquet_info("input.parquet")
# 测试 PostgreSQL 转换
result = convert_to_postgres(
parquet_path="input.parquet",
table_name="my_table"
)
# 测试 Markdown 处理
result = process_markdown_file(
file_path="input.md",
output_path="output.parquet"
)
✨ 主要特性
本项目提供了一系列处理 Parquet 文件的工具，涵盖嵌入、信息获取、格式转换等功能，具体如下：
- 嵌入工具：为指定列生成嵌入向量并保存到新列。
- Parquet 文件信息工具：获取 Parquet 文件的基本信息和统计信息。
- DuckDB 转换工具：将 Parquet 文件转换为 DuckDB 表格文件。
- PostgreSQL 转换工具：将 Parquet 文件数据导入 PostgreSQL 数据库表中。
- Markdown 处理工具：将 Markdown 文件转换为结构化的 Parquet 文件。
📦 安装指南
依赖安装
pip install -r requirements.txt
环境变量配置
在 .env 文件中添加必要的环境变量，如 OPENAI_API_KEY 等。
启动服务器
python src/run_server.py
💻 使用示例
基础用法
from parquet_mcp_server.client import (
convert_to_duckdb,
embed_parquet,
get_parquet_info,
convert_to_postgres,
process_markdown_file
)
# 测试 DuckDB 转换
result = convert_to_duckdb(
parquet_path="input.parquet",
output_dir="db_output"
)
# 测试嵌入功能
result = embed_parquet(
input_path="input.parquet",
output_path="output.parquet",
column_name="text",
embedding_column="embeddings",
batch_size=2
)
# 测试 Parquet 信息工具
result = get_parquet_info("input.parquet")
# 测试 PostgreSQL 转换
result = convert_to_postgres(
parquet_path="input.parquet",
table_name="my_table"
)
# 测试 Markdown 处理
result = process_markdown_file(
file_path="input.md",
output_path="output.parquet"
)
📚 详细文档
功能模块说明
嵌入工具
def embed_parquet(
input_path: str,
output_path: str,
column_name: str = "text",
embedding_column: str = "embeddings",
batch_size: int = 16
) -> None:
# 功能：为指定列生成嵌入向量并保存到新列
# 输入：
# input_path: 输入 Parquet 文件路径
# output_path: 输出 Parquet 文件路径
# column_name: 需要生成嵌入的列名，默认为 "text"
# embedding_column: 嵌入结果保存的列名，默认为 "embeddings"
# batch_size: 处理批次大小，默认为 16
Parquet 文件信息工具
def get_parquet_info(parquet_path: str) -> Dict:
# 功能：获取 Parquet 文件的基本信息和统计信息
# 输入：
# parquet_path: Parquet 文件路径
# 输出：
# Dict: 包含文件大小、行数、列信息等的字典
DuckDB 转换工具
def convert_to_duckdb(
parquet_path: str,
output_path: str,
table_name: str = "parquet_table"
) -> None:
# 功能：将 Parquet 文件转换为 DuckDB 表格文件
# 输入：
# parquet_path: 输入 Parquet 文件路径
# output_path: 输出 .db 文件路径
# table_name: 目标表名，默认为 "parquet_table"
PostgreSQL 转换工具
def convert_to_postgres(
parquet_path: str,
host: str,
database: str,
user: str,
password: str,
table_name: str = "parquet_table",
chunk_size: int = 1000
) -> None:
# 功能：将 Parquet 文件数据导入 PostgreSQL 数据库表中
# 输入：
# parquet_path: 输入 Parquet 文件路径
# host: 数据库主机地址
# database: 数据库名称
# user: 数据库用户名
# password: 数据库密码
# table_name: 目标表名，默认为 "parquet_table"
# chunk_size: 每次插入的数据量，默认为 1000
Markdown 处理工具
def process_markdown_file(
file_path: str,
output_path: str,
encoding: str = "utf-8",
split_by: str = None,
keep_separators: bool = False
) -> None:
# 功能：将 Markdown 文件转换为结构化的 Parquet 文件
# 输入：
# file_path: 输入 Markdown 文件路径
# output_path: 输出 Parquet 文件路径
# encoding: 字符编码，默认为 "utf-8"
# split_by: 按分隔符分割内容（可选）
# keep_separators: 是否保留分隔符，默认为 False
测试
启动测试服务器
python src/run_server.py # 启动 MCP 服务端
执行测试用例
python -m pytest tests/ # 运行所有测试用例
python -m pytest tests/test_embedding.py # 测试嵌入功能
python -m pytest tests/test_parquet_info.py # 测试 Parquet 文件信息工具
python -m pytest tests/test_duckdb_conversion.py # 测试 DuckDB 转换工具
python -m pytest tests/test_postgres_conversion.py # 测试 PostgreSQL 转换工具
python -m pytest tests/test_markdown_processing.py # 测试 Markdown 处理工具
API 响应格式
嵌入响应
{
"status": "success",
"message": "embedding completed successfully",
"result": {
"input_count": 100,
"output_file": "output.parquet"
}
}
文件信息响应
{
"file_size": "12.5 MB",
"num_rows": 1000,
"columns": [
{"name": "text", "type": "string"},
{"name": "label", "type": "int"}
]
}
🔧 技术细节
本项目使用 Python 语言开发，依赖于多个第三方库，如 llama-cpp-python、duckdb 等。在处理 Parquet 文件时，采用了分批处理的方式，以提高处理大文件时的性能。同时，通过设置环境变量的方式来管理 API 密钥等敏感信息，增强了项目的安全性。
📄 注意事项
⚠️ 重要提示
- 确保安装了所有依赖项，特别是
llama-cpp-python和duckdb。- 设置正确的环境变量以避免 API 调用错误。
- 处理大文件时建议分批处理以保证性能。