Distributed Ingest

1
2
3
4
5

from gpudb import GPUdb, GPUdbTable

db = GPUdb("http://<db.host>:9191", username = "<user>", password = "<pass>")

t = GPUdbTable(name = table_name, db = db, use_multihead_io = True)

1
2
3
4
5
6
7
8
9

from gpudb import GPUdb, GPUdbTable
from gpudb_multihead_io import GPUdbWorkerList, GPUdbIngestor

db = GPUdb("http://<db.host>:9191", username = "<user>", password = "<pass>")
workers = GPUdbWorkerList(db)

# Distributed Ingest Object Instantiation
table_type = GPUdbTable(name = table_name, db = db).get_table_type()
bulk_ingestor = GPUdbIngestor( db, table_name, table_type, batch_size, options, workers )

1

GPUdbWorkerList(db, "172.")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

// Acquire a type for the target table
final Type recordType = Type.fromTable(kdb, tableName);

// Construct a bulk inserter to perform the distributed ingest
try (BulkInserter<GenericRecord> bulkInserter = new BulkInserter<>(kdb, tableName, recordType, batchSize))
{
    // Insert records; any worker queues that are full after this list
    // is processed will be inserted in parallel
    bulkInserter.insert(records);

    // Insert any remaining records by flushing the BulkInserter object
    bulkInserter.flush();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

// Acquire a type for the target table
final Type recordType = Type.fromTable(kdb, tableName);

// Construct a bulk inserter to perform the distributed ingest
try (BulkInserter<GenericRecord> bulkInserter = new BulkInserter<>(kdb, tableName, recordType, batchSize))
{
    int recordNum = 0;

    // Insert records; worker queues should fill up, but never reach the
    // batch size and automatically flush.  With more than one queue and
    // an even distribution of records, by the time a batch-size number
    // of records has been processed, there should only be batchSize/n
    // records per queue, at which point a manual flush will be invoked.
    for (GenericRecord record : records)
    {
        bulkInserter.insert(record);
        
        // Insert records in parallel by flushing the BulkInserter object
        // every batch-size number of records
        if (++recordNum % batchSize == 0)
            bulkInserter.flush();
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

// Acquire a type for the target table
final Type recordType = Type.fromTable(kdb, tableName);

// Create timed flush options:  all queues, regardless of fullness, every second
BulkInserter.FlushOptions flushOptions = new BulkInserter.FlushOptions(false, 1);

// Construct a bulk inserter to perform the distributed ingest
try (BulkInserter<GenericRecord> bulkInserter =
        new BulkInserter<>(kdb, tableName, recordType, batchSize, null, null, flushOptions))
{
    // Insert records; no worker queues should fill up and automatically
    // flush, as the batch size equals the total record count.  This
    // will allow the timed flush to insert queued records every second.
    for (GenericRecord record : records)
        bulkInserter.insert(record);

    // Insert any remaining records by flushing the BulkInserter object
    bulkInserter.flush();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

// Acquire a type for the target table
final Type recordType = Type.fromTable(kdb, tableName);

// Construct a bulk inserter to perform the distributed ingest
try (BulkInserter<GenericRecord> bulkInserter = new BulkInserter<>(kdb, tableName, recordType, batchSize))
{
    ExecutorService executorService = null;

    try
    {
        // Create a thread pool for inserting data
        executorService = Executors.newFixedThreadPool(totalThreads);
    
        final int recordsPerThread = records.size() / totalThreads;

        for (int startIndex = 0; startIndex < records.size(); startIndex+=recordsPerThread)
        {
            // Pass the shared BulkInserter & record list into each task
            // with an offset for each to start inserting from
            executorService.execute
            (
                    new BatchInsert(bulkInserter, records, startIndex, recordsPerThread)
            );
        }
    }
    finally
    {
        // Shut down thread pool
        if (executorService != null)
        {
            executorService.shutdown();
            try
            {
                if (!executorService.awaitTermination(30, TimeUnit.SECONDS))
                    executorService.shutdownNow();
            }
            catch (@SuppressWarnings("unused") InterruptedException e)
            {
                executorService.shutdownNow();
            }
        }
    }

    // Insert any remaining records by flushing the BulkInserter object
    bulkInserter.flush();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

/**
 * Helper class for parallelizing usage of the bulk ingest object.
 * 
 * This class will generate a batch of inserts of a given size, starting at
 * a given index, and then add them to a given BulkInserter
 */
private static class BatchInsert implements Runnable
{
    /* Bulk ingestion object to use for inserting records*/
    BulkInserter<GenericRecord> bulkInserter;
    /* Full set of records being inserted */
    List<GenericRecord> records;
    /* ID/index of first record to insert within the full set */
    int startIndex;
    /* ID/index of last record to insert within the full set */
    int endIndex;

    /*
     * Creates a new batch insert object, which will generate objects of the
     * specified type and add them to the specified bulk ingest object
     * 
     * @param bulkInserter bulk ingest object to which records will be added
     * @param type database type schema to use for objects being inserted
     * @param startIndex starting index of records being inserted; will be
     *        used as order ID for inserted records
     * @param batchSize number of records to insert in this batch
     */
    public BatchInsert(BulkInserter<GenericRecord> bulkInserter, List<GenericRecord> records, int startIndex, int recordCount)
    {
        this.bulkInserter = bulkInserter;
        this.records = records;
        this.startIndex = startIndex;
        this.endIndex = startIndex + recordCount;
    }

    public void run()
    {
        try
        {
            for (int recIndex = this.startIndex; recIndex < this.endIndex; recIndex++)
            {
                this.bulkInserter.insert(this.records.get(recIndex));
            }
        }
        catch (Exception e)
        {
            System.out.println("Error inserting record: " + e);
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

/**
 * Create a class whose members map, by name and type, to the target
 * table it represents.  Annotate each column with its ordering in the
 * target table and any column properties.
 */
public static class Product extends RecordObject
{
    @RecordObject.Column(order = 0, properties = { "primary_key" })
    public Integer id;
    @RecordObject.Column(order = 1, properties = { "char64" })
    public String name;
    @RecordObject.Column(order = 2)
    public String description;
    @RecordObject.Column(order = 3)
    public Long ts;

    public Product() {}

    /* Create a constructor for the class that will take parameters so that
     * assigning values is easier */
    public Product(Integer id, String name, String description)
    {
        this.id = id;
        this.name = name;
        this.description = description;
        this.ts = Long.MIN_VALUE;  // Invalid timestamp to trigger INIT_WITH_NOW
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

// Acquire a type for the order history table
final Type recordType = Type.fromTable(kdb, tableName);

// Construct a bulk inserter from the table type to perform the distributed ingest
try (BulkInserter<GenericRecord> bulkInserter =
        new BulkInserter<>(kdb, tableName, recordType, batchSize))
{
    for (Map<String, Object> product : products)
    {
        // Create record from the table type and populate
        GenericRecord record = new GenericRecord(recordType);
        record.put(0, product.get("id"));
        record.put(1, product.get("name"));
        record.put(2, product.get("description"));
        record.put(3, Long.MIN_VALUE);  // Invalid timestamp to trigger INIT_WITH_NOW
        bulkInserter.insert(record);
    }

    // Insert any remaining records by flushing the BulkInserter object
    bulkInserter.flush();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// Construct a bulk inserter with no type specified to perform JSON distributed ingest
try (BulkInserter<String> bulkInserter = new BulkInserter<>(kdb, tableName, batchSize))
{
    // Insert records as JSON strings
    for (String product : products)
        bulkInserter.insert(product);

    // Insert any remaining records by flushing the BulkInserter object
    bulkInserter.flush();
}