Julia 客户端 – DuckDB - DuckDB 数据库

搜索快捷键 cmd + k | ctrl + k

文档 / 客户端 API

Julia 客户端

DuckDB Julia 程序包为 DuckDB 提供了高性能的前端。与 SQLite 类似，DuckDB 在 Julia 客户端内以进程内（in-process）方式运行，并提供了一个 DBInterface 前端。

该程序包还支持多线程执行。它为此使用 Julia 的线程/任务。如果您希望并行运行查询，则必须在启用多线程支持的情况下启动 Julia（例如，通过设置 JULIA_NUM_THREADS 环境变量）。

安装

安装 DuckDB 的方法如下

using Pkg
Pkg.add("DuckDB")

或者，使用 ] 键进入程序包管理器，并输入以下命令

pkg> add DuckDB

基础知识

using DuckDB

# create a new in-memory database
con = DBInterface.connect(DuckDB.DB, ":memory:")

# create a table
DBInterface.execute(con, "CREATE TABLE integers (i INTEGER)")

# insert data by executing a prepared statement
stmt = DBInterface.prepare(con, "INSERT INTO integers VALUES (?)")
DBInterface.execute(stmt, [42])

# query the database
results = DBInterface.execute(con, "SELECT 42 a")
print(results)

某些 SQL 语句（如 PIVOT 和 IMPORT DATABASE）会作为多个预处理语句执行，在使用 DuckDB.execute() 时会报错。对于这些语句，可以使用 DuckDB.query() 代替 DuckDB.execute() 来运行，它们将始终返回物化结果。

扫描 DataFrames

DuckDB Julia 程序包还支持查询 Julia DataFrames。请注意，DataFrame 是由 DuckDB 直接读取的——它们不会被插入或复制到数据库本身中。

如果您希望将数据从 DataFrame 加载到 DuckDB 表中，可以运行 CREATE TABLE ... AS 或 INSERT INTO 查询。

using DuckDB
using DataFrames

# create a new in-memory database
con = DBInterface.connect(DuckDB.DB)

# create a DataFrame
df = DataFrame(a = [1, 2, 3], b = [42, 84, 42])

# register it as a view in the database
DuckDB.register_data_frame(con, df, "my_df")

# run a SQL query over the DataFrame
results = DBInterface.execute(con, "SELECT * FROM my_df")
print(results)

Appender API

DuckDB Julia 程序包还支持 Appender API，它比使用预处理语句或单独的 INSERT INTO 语句要快得多。追加操作以行格式进行。对于每一列，应调用一次 append()，之后应调用 flush() 来结束该行。在所有行都追加完成后，应使用 close() 来完成 Appender 的操作并清理产生的内存。

using DuckDB, DataFrames, Dates
db = DuckDB.DB()
# create a table
DBInterface.execute(db,
    "CREATE OR REPLACE TABLE data (id INTEGER PRIMARY KEY, value FLOAT, timestamp TIMESTAMP, date DATE)")
# create data to insert
len = 100
df = DataFrames.DataFrame(
        id = collect(1:len),
        value = rand(len),
        timestamp = Dates.now() + Dates.Second.(1:len),
        date = Dates.today() + Dates.Day.(1:len)
    )
# append data by row
appender = DuckDB.Appender(db, "data")
for i in eachrow(df)
    for j in i
        DuckDB.append(appender, j)
    end
    DuckDB.end_row(appender)
end
# close the appender after all rows
DuckDB.close(appender)

并发

在 Julia 进程中，任务可以并发地对数据库进行读写操作，前提是每个任务都维护其自己的数据库连接。在下面的示例中，生成了一个任务用于定期读取数据库，并生成了许多任务通过 INSERT 语句以及 Appender API 向数据库写入数据。

using Dates, DataFrames, DuckDB
db = DuckDB.DB()
DBInterface.connect(db)
DBInterface.execute(db, "CREATE OR REPLACE TABLE data (date TIMESTAMP, id INTEGER)")

function run_reader(db)
    # create a DuckDB connection specifically for this task
    conn = DBInterface.connect(db)
    while true
        println(DBInterface.execute(conn,
                "SELECT id, count(date) AS count, max(date) AS max_date
                FROM data GROUP BY id ORDER BY id") |> DataFrames.DataFrame)
        Threads.sleep(1)
    end
    DBInterface.close(conn)
end
# spawn one reader task
Threads.@spawn run_reader(db)

function run_inserter(db, id)
    # create a DuckDB connection specifically for this task
    conn = DBInterface.connect(db)
    for i in 1:1000
        Threads.sleep(0.01)
        DuckDB.execute(conn, "INSERT INTO data VALUES (current_timestamp, ?)"; id);
    end
    DBInterface.close(conn)
end
# spawn many insert tasks
for i in 1:100
    Threads.@spawn run_inserter(db, 1)
end

function run_appender(db, id)
    # create a DuckDB connection specifically for this task
    appender = DuckDB.Appender(db, "data")
    for i in 1:1000
        Threads.sleep(0.01)
        row = (Dates.now(Dates.UTC), id)
        for j in row
            DuckDB.append(appender, j);
        end
        DuckDB.end_row(appender);
    end
    DuckDB.close(appender);
end
# spawn many appender tasks
for i in 1:100
    Threads.@spawn run_appender(db, 2)
end

原始 Julia 连接器

感谢 kimmolinna 提供的原始 DuckDB Julia 连接器。

关于此页面

行为准则商标使用指南

安装

基础知识

扫描 DataFrames

Appender API

并发

原始 Julia 连接器

关于此页面

本文中