Пакет Julia для DuckDB

Пакет DuckDB Julia предоставляет высокопроизводительный фронтенд для DuckDB. Подобно SQLite, DuckDB работает в процессе внутри клиента Julia и предоставляет фронтенд 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. Обратите внимание, что DataFrames считываются непосредственно DuckDB — они не вставляются и не копируются в саму базу данных.

Если вы хотите загрузить данные из DataFrame в таблицу DuckDB, вы можете запустить CREATE TABLE ... AS или INSERT INTO запрос.

using DuckDB
using DataFrames

# create a new in-memory dabase
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)

API аппендера

Пакет DuckDB Julia также поддерживает API аппендера, который намного быстрее, чем использование подготовленных запросов или отдельных INSERT INTO запросов. Дополнения выполняются в строчном формате. Для каждого столбца необходимо вызвать append(), после чего строку следует завершить, вызвав flush(). После добавления всех строк необходимо использовать close(), чтобы завершить работу аппендера и очистить полученную память.

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 запросы, так и 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.