nbraun/dask-sql

dask-sql

镜像概述和主要用途

dask-sql是一个在dask之上添加SQL查询层的工具，允许用户通过常见SQL操作查询和转换Dask数据帧。其查询以标准Dask操作方式运行，可在Dask集群中分布式执行。该工具目标类似于Hadoop生态中的Spark SQL/Hive/Drill等，但目前功能较精简。

核心功能和特性

• SQL查询支持：通过SQL语法操作Dask数据帧，支持常见SQL操作（如SELECT、JOIN、GROUP BY等）
• 分布式执行：查询作为Dask操作运行，可在Dask集群中分布式处理
• Dask生态集成：与Dask数据帧无缝集成，查询结果为Dask数据帧，可继续进行其他Dask计算
• SQL服务器：提供基于Presto协议的SQL服务器，支持Presto客户端连接
• 多环境支持：可通过conda、pip安装，或使用Docker镜像快速部署

使用场景和适用范围

• 数据分析：通过SQL对Dask数据帧进行查询和分析
• 数据转换：使用SQL进行数据清洗、过滤和转换
• 分布式查询：在Dask集群中执行SQL查询，处理大规模数据集
• 快速原型验证：通过Docker镜像快速启动SQL服务器，测试SQL查询逻辑

使用方法和配置说明

Docker快速部署

通过以下命令启动dask-sql SQL服务器：

bash
docker run --rm -it -p 8080:8080 nbraun/dask-sql

该命令会在本地8080端口启动一个SQL服务器，支持Presto客户端连接。

SQL服务器使用

1. 连接服务器：使用Presto客户端连接到本地服务器

   bash
   presto --server localhost:8080
   

2. 执行查询：在Presto客户端中执行SQL查询

   sql
   => SELECT 1 + 1;
    EXPR$0
   --------
       2
   (1 row)
   

Python API使用示例

python
from dask_sql import Context
from dask.datasets import timeseries

# 创建上下文以保存已注册的表
c = Context()

# 加载数据并注册为表
df = timeseries()
c.create_table("timeseries", df)

# 执行SQL查询，结果为Dask数据帧
result = c.sql("""
    SELECT
        lhs.name,
        lhs.id,
        lhs.x
    FROM
        timeseries AS lhs
    JOIN
        (
            SELECT
                name AS max_name,
                MAX(x) AS max_x
            FROM timeseries
            GROUP BY name
        ) AS rhs
    ON
        lhs.name = rhs.max_name AND
        lhs.x = rhs.max_x
""")

# 计算并显示结果
print(result.compute())

# 进一步计算（示例）
print(result.x.mean().compute())

其他安装方式

通过conda安装（推荐）

bash
conda create -n dask-sql
conda activate dask-sql
conda install dask-sql -c conda-forge

通过pip安装

需先安装Java 8+，验证Java安装：

bash
java -version

然后安装包：

bash
pip install dask-sql

工作原理

dask-sql主要完成两个核心任务：

1. 使用Apache Calcite将SQL查询转换为关系代数（以Java对象树表示）
2. 将关系代数转换为Dask API调用，生成并执行Dask执行计划，返回Dask数据帧

注意事项

• dask-sql目前处于开发阶段，尚未支持所有SQL命令
• 欢迎反馈、改进和贡献代码

更多信息请参阅官方文档