Creating Tables & Writing Data

Vulkn allows you to create and append to tables as you would any other data system.

Each DataTable object supports creating new tables and appending to existing tables.

write(database, table, engine=None, mode=WriteMode.Create)

  • Parameters
    • database: str - the database to write the table to
    • table: str - the tablename to write to
    • engine: vulkn.engines.* - the engine configuration to use for the table (default None)
    • mode: vulkn.datatable.WriteMode - the strategy to use when writing the data - whether to Create, Append or Replace the table (if it exists).

Writes/creates or appends the DataTable to a table with the specified engine.

Available WriteModes

Valid WriteModes (vulkn.datatable.WriteMode) are:

  • WriteMode.Create - Create the table if it doesn't exist
  • WriteMode.Append - Append to the table if it does exist (will result in create otherwise)
  • WriteMode.Replace - Replace the table (overwrite) if it already exists


The following two code blocks are effectively identical:

DROP TABLE IF EXISTS data.agg_events;

CREATE TABLE data.agg_events 
ENGINE = MergeTree ORDER BY (geo_id)
        uniqExact(event_msg) AS total_uniq_messages,
        sum(event_metric) AS sum_event_metrics
    GROUP BY geo_id;

SELECT count() FROM data.agg_events;
import vulkn.engines as engines
from vulkn.datatable import WriteMode

df = (v.table('')
    .write('data', 'agg_events',
           engine=engines.MergeTree(order_by=('geo_id',)), WriteMode.Replace))


Efficiencies, code re-use and maintainability become apparent in more complex examples.

In the next example we first define a table engine strategy/profile. We then define a template query for a very large table,, that we wish to join/denormalize with a dimension table. Because the tables are very large, and we can't hold all the data in memory, we iterate through all unique geo_id dimensions and append the results to the events.denomalized_events table.

The Append WriteMode will Create the table on first call and append to the table on subsequent calls. If we had more than a handful of geo_ids the SQL-only version becomes unfeasible.

import vulkn.engines as engines
from vulkn.datatable import WriteMode

table_profile = engines.MergeTree(partition_by=('toDate(event_date)',),

events = v.q('select * from').where("geo_id == {geo_id:UInt32}").alias('e')
dimensions = v.table('default.dims').select('geo_id', 'geo_location', 'geo_coords').alias('d')

for g in v.table('default.dims').select('geo_id').distinct().r:
    df = (events
        .lj(dimensions, ('geo_id',))
        .select('d.*', 'e.*')
        .params({'geo_id': g['geo_id']})
        .write('events', 'denormalized_events', engine=table_profile, WriteMode.Append))

    print(f'events.denormalized_events total count is now {df.count().l[0][0]}')