Julia-Client

Das DuckDB-Julia-Paket bietet ein leistungsstarkes Front-End für DuckDB. Ähnlich wie SQLite läuft DuckDB im Prozess innerhalb des Julia-Clients und stellt ein DBInterface-Front-End bereit.

Das Paket unterstützt außerdem die mehrthreadige Ausführung. Dazu verwendet es Julia-Threads/Tasks. Wenn Sie Abfragen parallel ausführen möchten, müssen Sie Julia mit Unterstützung für Multithreading starten (z. B. indem Sie die Umgebungsvariable JULIA_NUM_THREADS setzen).

Installation

Installieren Sie DuckDB wie folgt:

using Pkg
Pkg.add("DuckDB")

Alternativ öffnen Sie den Paketmanager mit der Taste ] und führen Sie den folgenden Befehl aus:

pkg> add DuckDB

Grundlagen

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)

Einige SQL-Anweisungen wie PIVOT und IMPORT DATABASE werden als mehrere Prepared Statements ausgeführt und führen bei Verwendung von DuckDB.execute() zu einem Fehler. Stattdessen können sie mit DuckDB.query() statt DuckDB.execute() ausgeführt werden und geben immer ein materialisiertes Ergebnis zurück.

DataFrames scannen

Das DuckDB-Julia-Paket unterstützt außerdem das Abfragen von Julia-DataFrames. Beachten Sie, dass die DataFrames direkt von DuckDB gelesen werden – sie werden nicht in die Datenbank selbst eingefügt oder kopiert.

Wenn Sie Daten aus einem DataFrame in eine DuckDB-Tabelle laden möchten, können Sie eine Abfrage CREATE TABLE ... AS oder INSERT INTO ausführen.

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

Das DuckDB-Julia-Paket unterstützt außerdem die Appender-API, die deutlich schneller ist als Prepared Statements oder einzelne INSERT INTO-Anweisungen. Das Anhängen erfolgt im zeilenweisen Format. Für jede Spalte sollte ein Aufruf append() erfolgen, danach sollte die Zeile mit flush() abgeschlossen werden. Nachdem alle Zeilen angehängt wurden, sollte close() verwendet werden, um den Appender abzuschließen und den entstehenden Speicher freizugeben.

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)

Nebenläufigkeit

Innerhalb eines Julia-Prozesses können Tasks gleichzeitig aus der Datenbank lesen und in sie schreiben, solange jeder Task seine eigene Verbindung zur Datenbank hält. Im folgenden Beispiel wird ein einzelner Task gestartet, der periodisch aus der Datenbank liest, und viele Tasks werden gestartet, um mit INSERT-Anweisungen sowie der Appender-API in die Datenbank zu schreiben.

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

Ursprünglicher Julia-Connector

Dank an kimmolinna für den ursprünglichen DuckDB-Julia-Connector.