MultithreadedTableWriter

The MultithreadedTableWriter (MTW) class collects single inserted records, batches them together, and asynchronously writes the batches to the server using multiple threads. When constructing an MTW object, you must specify the number of writer threads (threadCount). MTW creates a total of threadCount+1 threads:

• threadCount writer threads
• 1 converter thread

This is how MTW works internally:

(1) The insert method accepts records (Python objects) one at a time and adds them to a converter queue. It returns an ErrorCodeInfo object. MTW only checks if the incoming data has the correct column count for the target table. The data types are not validated. If an error (e.g., incorrect column count, writer thread has quit) occurs during insert(), ErrorCodeInfo.hasError() returns True.

(2) During construction of the MTW object, a converter thread is spawned to convert the incoming record (Python object) into a C++ object, decreasing the performance impact of the Python Global Interpreter Lock (GIL). If an error occurs, all background threads including the converter thread will terminate, and the error will be logged to MultithreadedWriterStatus, which you can obtain via MultithreadedTableWriter.getStatus(). The converted C++ objects are distributed to the writer threads, and the Python objects that failed conversion can be accessed by calling MultithreadedTableWriter.getUnwrittenData().

(3) The threadCount writer threads transmit the data and associated script to the DolphinDB server according to the specified batchSize and throttle parameters. As with the previous step, if an error occurs during this process, all background threads will terminate and the error will be logged to the MultithreadedWriterStatus.

(4) The waitForThreadCompletion() method of the MTW class will wait in a blocking manner for all writing tasks to complete, similar to calling .join() on the internal threads.

**Note:** After this method has been called, the MTW object can no longer be used to insert additional data, as it signifies all writing operations have ended. A new MTW object must be created to continue data insertion.

MultithreadedTableWriter(host, port, userId, password, dbPath, tableName, useSSL=False, enableHighAvailability=False, highAvailabilitySites=[], batchSize=1, throttle=1, threadCount=1, partitionCol="", compressMethods=[], mode="", modeOption=[])

The methods of MultithreadedTableWriter are usually used in the following order:

# prepare data
prepre_Data()
# Construct an MTW object
writer = ddb.MultithreadedTableWriter(...)

try:
    for data in datas:
        # insert data
        res = writer.insert(data)
        # check if data has been added to the conversion queue
        if res.hasError():
            print(res.errorInfo)
    # wait for MTW to complete
    writer.waitForThreadCompletion()
except Exception as e:
    # Get the status of MTW and check for errors
    writeStatus = writer.getStatus()
    print(e)
    if writeStatus.hasError():
        # obtain unwritten data, fix the issues, and insert again
        print(writeStatus)
        unwrittendata = writer.getUnwrittenData()
        # fix the unwritten data with a predefined function
        unwrittendata = revise(unwrittendata)
        newwriter = ddb.MultithreadedTableWriter(...)
        newwriter.insertUnwrittenData(unwrittendata)

This example uses the MTW to write data to a DolphinDB DFS partitioned table.

(1) Create a DolphinDB DFS partitioned table

import numpy as np
import pandas as pd
import dolphindb as ddb
import time
import random

s = ddb.Session()
s.connect("localhost", 8848, "admin", "123456")

script = """
    dbName = 'dfs://valuedb3';
    if(exists(dbName)){
        dropDatabase(dbName);
    }
    datetest=table(1000:0,`date`symbol`id,[DATE,SYMBOL,INT]);
    db = database(directory=dbName, partitionType=HASH, partitionScheme=[INT, 10]);
    pt=db.createPartitionedTable(datetest,'pdatetest','id');
"""
s.run(script)

(2) Create an MTW object

writer = ddb.MultithreadedTableWriter(
    "localhost", 8848, "admin", "123456", dbPath="dfs://valuedb3", tableName="pdatetest", batchSize=10000, throttle=1, threadCount=5, partitionCol="id", compressMethods=["LZ4","LZ4","DELTA"]
)

In this step, we create an MTW object with 5 threads to write data to the partitioned table "pdatetest" under DFS database "dfs://valuedb3". The batchSize, throttle, partitionCol, and the compression method for each column are specified.

(3) Write data

try:
    # insert 100 records with correct schema
    for i in range(100):
        res = writer.insert(random.randint(1,10000),"AAAAAAAB", random.randint(1,10000))
        if res.hasError():
            print("MTW insert error: ", res.errorInfo)
except Exception as ex:
    # exception handling
    print("MTW exited with exception: ", ex)
# wait for completion
writer.waitForThreadCompletion()
writeStatus = writer.getStatus()
if writeStatus.succeed():
    print("Write successful!")
print("writeStatus: \n", writeStatus)
print(s.run("select count(*) from pt"))

Call writer.insert() insert data and writer.getStatus() to check the status of writer.

Output

Write successful!
writeStatus: 
 errorCode     : None
 errorInfo     : 
 isExiting     : True
 sentRows      : 100
 unsentRows    : 0
 sendFailedRows: 0
 threadStatus  : 
        threadId        sentRows        unsentRows      sendFailedRows
               0               0                 0                   0
        1677719296            22                 0                   0
        1669326592            20                 0                   0
        1660933888            19                 0                   0
        1652541184            14                 0                   0
        1644148480            25                 0                   0
<dolphindb.session.MultithreadedTableWriterStatus object at 0x7fe896260d30>
   count
0    100

The output shows that all data has been successfully written to the DFS table. The errorCode is "None".

Error Handling

The following example demonstrates how to catch exceptions with getStatus and try-catch during data writes.

Note

(1) The DFS table already has 100 records from the previous steps. (2) As we have called waitForCompletion in step 3, the previous thread has exited. Now we need to construct a new MTW object to continue data insertion.

writer = ddb.MultithreadedTableWriter(
    "localhost", 8848, "admin", "123456", dbPath="dfs://valuedb3", tableName="pdatetest",
    batchSize=10000, throttle=1, threadCount=5, partitionCol="id", compressMethods=["LZ4","LZ4","DELTA"]
)

try:
    # insert 100 records with the correct data types and column count
    for i in range(100):
        res = writer.insert(np.datetime64('2022-03-23'),"AAAAAAAB", random.randint(1,10000))
    # Insert 10 records with incorrect data types. As no type check is performed at this stage, these records will be added to the converter queue
    # The incorrect data types will be detected only when conversion starts. Then all background threads of MTW will be terminated.
    for i in range(10):
        res = writer.insert(np.datetime64('2022-03-23'),222, random.randint(1,10000))
        if res.hasError():
            # no error will be detected at this stage
            print("Insert wrong format data:\n", res)
    # Insert a record with incorrect column count. An error will be raised immediately
    # This record will not be added to converter queue
    res = writer.insert(np.datetime64('2022-03-23'),"AAAAAAAB")
    if res.hasError():
        # print error message
        print("Column counts don't match:\n", res)
    # Sleep for 1 second to wait for the conversion starts. The 10 records with incorrect data types will be detected.  
    # At this stage, all threads are terminated. MTW status is changed accordingly. 
    time.sleep(1)

    # # insert 1 record of correct type and column count. As MTW has exited, the record is not inserted.
    res = writer.insert(np.datetime64('2022-03-23'),"AAAAAAAB", random.randint(1,10000))
    print("MTW has exited")
except Exception as ex:
    # exception message
    print("MTW exited with exception %s" % ex)
# wait for completion
writer.waitForThreadCompletion()
writeStatus = writer.getStatus()
if writeStatus.hasError():
    print("Error in writing:")
print(writeStatus)
print(s.run("select count(*) from pt"))
"""
Column counts don't match:
 errorCode: A2
 errorInfo: Column counts don't match 3
<dolphindb.session.ErrorCodeInfo object at 0x7f0d52947040>
MTW exited with exception <Exception> in insert: thread is exiting.
Error in writing:
errorCode     : A1
 errorInfo     : Data conversion error: Cannot convert long to SYMBOL
 isExiting     : True
 sentRows      : 0
 unsentRows    : 4
 sendFailedRows: 106
 threadStatus  : 
        threadId        sentRows        unsentRows      sendFailedRows
               0               0                 0                 106
        511690496              0                 2                   0
        520083200              0                 1                   0
        528475904              0                 0                   0
        618751744              0                 0                   0
        536868608              0                 1                   0
<dolphindb.session.MultithreadedTableWriterStatus object at 0x7f0d25d68910>
   count
0    100
"""

** Re-insert**

When the MTW encounters an error during data conversion or insertion, it will terminate all threads. In this situation, you can call writer.getUnwrittenData() to retrieve the data that was not inserted, then call insertUnwrittenData(unwrittendata) to re-attempt inserting the unwritten data. Note that since the original MTW has terminated its threads, a new MTW must be created before the unwritten data can be inserted again.

if writeStatus.hasError():
    print("Error in writing:")
    unwrittendata = writer.getUnwrittenData()
    print("Unwrittendata: %d" % len(unwrittendata))
    # construct a new MTW
    newwriter = ddb.MultithreadedTableWriter(
        "localhost", 8848, "admin", "123456", dbPath="dfs://valuedb3", tableName="pdatetest",
        batchSize=10000, throttle=1, threadCount=5, partitionCol="id", compressMethods=["LZ4","LZ4","DELTA"]
    )
    try:
        # fix the issue with the unwritten data and insert again
        for row in unwrittendata:
            row[1]="aaaaa"
        res = newwriter.insertUnwrittenData(unwrittendata)
        if res.hasError():
            print("Failed to write data again: \n", res)
    except Exception as ex:
        # print error message
        print("MTW exit with exception %s" % ex)
    finally:
        # wait for newwriter to complete
        newwriter.waitForThreadCompletion()
        writeStatus = newwriter.getStatus()
        print("Write again:\n", writeStatus)
else:
    print("Write successfully:\n", writeStatus)

print(s.run("select count(*) from pt"))
"""
Error in writing:
Unwrittendata: 110
Write again:
 errorCode     : None
 errorInfo     : 
 isExiting     : True
 sentRows      : 110
 unsentRows    : 0
 sendFailedRows: 0
 threadStatus  : 
        threadId        sentRows        unsentRows      sendFailedRows
               0               0                 0                   0
        618751744             17                 0                   0
        528475904             21                 0                   0
        520083200             24                 0                   0
        511690496             25                 0                   0
        503297792             23                 0                   0
<dolphindb.session.MultithreadedTableWriterStatus object at 0x7f0d52947040>
   count
0    210
"""

The DolphinDB Python API ensures thread safety of Python multi-threaded call of MTW. For example:

import numpy as np
import pandas as pd
import dolphindb as ddb
import time
import threading
import random

s = ddb.Session()
s.connect("localhost", 8848, "admin", "123456")

script = """
    dbName = 'dfs://valuedb3';
    if(exists(dbName)){
        dropDatabase(dbName);
    }
    datetest=table(1000:0,`date`symbol`id,[DATE,SYMBOL,INT]);
    db = database(directory=dbName, partitionType=HASH, partitionScheme=[INT, 10]);
    pt=db.createPartitionedTable(datetest,'pdatetest','id');
"""
s.run(script)

writer = ddb.MultithreadedTableWriter(
    "localhost", 8848, "admin", "123456", dbPath="dfs://valuedb3", tableName="pdatetest",
    batchSize=10000, throttle=1, threadCount=5, partitionCol="id", compressMethods=["LZ4","LZ4","DELTA"]
)

def insert_MTW(writer):
    try:
        # insert 100 records of correct data types and schema
        for i in range(100):
            res = writer.insert(random.randint(1,10000),"AAAAAAAB", random.randint(1,10000))
    except Exception as ex:
        # print error message
        print("MTW exit with exception %s" % ex)

# create 10 threads to insert data into MTW
threads = []
for i in range(10):
    threads.append(threading.Thread(target=insert_MTW, args=(writer,)))

for thread in threads:
    thread.start()

# execute other tasks while the threads insert data. Sleeping for 10 seconds in this example.
time.sleep(10)

# finish the task
# 1 - wait for threads to quit
for thread in threads:
    thread.join()
# 2 - wait for MTW to complete
writer.waitForThreadCompletion()
# 3 - check insertion result
writeStatus = writer.getStatus()
print("writeStatus:\n", writeStatus)
print(s.run("select count(*) from pt"))

Output:

writeStatus:
 errorCode     : None
 errorInfo     : 
 isExiting     : True
 sentRows      : 1000
 unsentRows    : 0
 sendFailedRows: 0
 threadStatus  : 
        threadId        sentRows        unsentRows      sendFailedRows
               0               0                 0                   0
        1309443840           194                 0                   0
        1301051136           188                 0                   0
        1292658432           215                 0                   0
        1284265728           208                 0                   0
        1275873024           195                 0                   0
<dolphindb.session.MultithreadedTableWriterStatus object at 0x7ff2802bf9a0>
   count
0   1000

This example demonstrates using the MTW to insert data with multiple Python threads. In this example, 10 Python threads were started and each thread inserts 100 records into the MTW. A total of 1000 records (10 threads x 100 records) were successfully written to the table through the MTW.