MultithreadedTableWriter

• host ：字符串，必填，表示所连接的服务器的 IP 地址。
• port ：整数，必填，表示所连接的服务器的端口号。
• userId ：字符串，可选，表示登录时的用户名。
• password ：字符串，可选，表示登录时用户名对应的密码。
• dbPath ：字符串，必填，表示分布式数据库地址。内存表时该参数为空。请注意，若使用 1.30.17.4 及以下版本的 API 向内存表写入数据时，该参数须填写内存表表名。
• tableName ：字符串，必填，表示分布式表或内存表的表名。请注意，若使用 1.30.17.4 及以下版本的 API 向内存表写入数据时，该参数须为空。
• useSSL ：布尔值，可选，默认值为 False。表示是否启用加密通讯。和 session 类似，当 MultithreadedTableWriter 启用加密通讯时，DolphinDB 需配置 enableHTTPS=true。
• enableHighAvailability / highAvailabilitySites ：高可用配置参数。有关配置方式，请参考章节 Connect。
• reconnect：布尔值，可选，默认值为 True。表示在写入过程中发生网络连接中断时是否自动重连。

• batchSize ：整数，表示批处理的消息的数量，默认值是1，表示客户端写入数据后就立即发送给服务器。如果该参数大于 1，表示数据量达到 batchSize 时，客户端才会将数据发送给服务器。
• throttle ：大于0的浮点数，单位为秒。若客户端有数据写入，但数据量小于 batchSize，则等待 throttle 的时间再发送数据。
• threadCount ：整数，表示创建的工作线程数量，默认为1，表示单线程。若写入对象为不支持并发写入的表，则该参数值必须为 1
• partitionCol ：字符串类型，默认为空，仅在 threadCount 大于1 时起效。若写入对象为分区表，必须指定任意一个分区字段；其他情况需指定为表的字段名。
• compressMethods ：列表，用于指定每一列采用的压缩传输方式，若该参数为空则表示不压缩。每一列可选的压缩方式（大小写不敏感）包括：
  • "LZ4"： LZ4 压缩。
  • "DELTA"： DELTAOFDELTA 压缩。
• mode ：字符串，表示数据写入的方式，可选值为 "Upsert" 或 "Append"。"Upsert" 表示以 upsert!方式更新表数据；"Append" 表示以tableInsert方式更新表数据。
• modeOption：字符串列表，表示upsert!可选参数组成的 list，仅当 mode 指定为 "Upsert" 时有效。

writer = ddb.MultithreadedTableWriter(
    "localhost", 8848, "admin", "123456", "dfs://testMTW", "pt", batchSize=2, throttle=0.01,  
    threadCount=3, partitionCol="date", compressMethods=["LZ4", "DELTA", "LZ4", "DELTA"],
    mode="Upsert", modeOption=["true", "`index"]
)

以上示例展示了如何构造一个 MultithreadedTableWriter，其中设置了3个写入线程，用于写入数据库 dfs://testMTW 中的分区表 pt。设置参数 batchSize=2、throttle=0.01，一旦当前待写入队列的长度超过2或者等待时间超过0.01秒，就会立即触发写入线程，将写入队列中的所有数据发送到 DolphinDB。其中 partitionCol 被指定为 date 列，当数据分配到待写入队列时，将根据 date 列将数据分别分配到三个写入线程对应的队列中。由于待写入表有四列数据，分别指定了压缩方式为 LZ4、DELTA、LZ4、DELTA。写入模式被指定为 Upsert，并传入了upsert!函数对应的参数 ["true", "`index"]，即每次写入时调用的 upsert! 函数的 ignoreNull 参数为 true，keyColNames 为 `index。

res:ErrorCodeInfo = writer.insert(*args)

data:list = writer.getUnwrittenData()

res:ErrorCodeInfo = insertUnwrittenData(unwrittenData)

status:MultithreadedTableWriterStatus = writer.getStatus()

writer.waitForThreadCompletion()

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)

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

在这一步中，创建一个包含 5 个线程的 MTW 对象，同时设置 batchSize=10000，throttle=1，指定分区列为 "id"、各列压缩方式分别为 LZ4、LZ4、DELTA，向 dfs://valuedb3 分布式数据库中的分区表 pdatetest 写入数据。

try:
    # 插入100行正确数据 
    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:
    # MTW 抛出异常
    print("MTW exit with exception: ", ex)
# 等待 MTW 插入完成
writer.waitForThreadCompletion()
writeStatus = writer.getStatus()
if writeStatus.succeed():
    print("Write successfully!")
print("writeStatus: \n", writeStatus)
print(s.run("select count(*) from pt"))

调用 writer.insert() 方法向 writer 中写入数据，并通过 writer.getStatus() 获取 writer 的状态。

Write successfully!
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

由上述结果可知，MultithreadedTableWriter 对象中的所有数据均成功写入分布式表，此时 errorCode 的输出为 "None"。

下述示例将介绍如果写入过程中发生错误，该如何通过 getStatus 和 try-catch 进行错误处理。

注意 ：

• 此时分区表中已有 100 条记录。
• 因为步骤三中 MTW 已经调用 waitForThreadCompletion 方法结束后台线程，所以在后续代码中需要重新构造一个新的 MTW 对象，才能继续写入数据。

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:
    # 插入100行正确数据 （类型和列数都正确），MTW正常运行
    for i in range(100):
        res = writer.insert(np.datetime64('2022-03-23'),"AAAAAAAB", random.randint(1,10000))
    # 插入10行类型错误数据，此时 MTW 并不会进行类型判断，这些数据能够进入 MTW 待转换队列
    # 直到转换线程对这些数据进行转换时，检测到类型不匹配，就会立刻终止 MTW 所有后台线程
    for i in range(10):
        res = writer.insert(np.datetime64('2022-03-23'),222, random.randint(1,10000))
        if res.hasError():
            # 此处不会执行到
            print("Insert wrong format data:\n", res)
    # 插入1行数据(列数不匹配)，MTW 立刻发现待插入数据列数与待插入表的列数不匹配，立刻返回错误信息，
    # 本条数据并不会进入待转换队列
    res = writer.insert(np.datetime64('2022-03-23'),"AAAAAAAB")
    if res.hasError():
        # 数据错误，插入列数不匹配数据
        print("Column counts don't match:\n", res)
    # sleep 1秒，等待 MTW 转换线程处理数据直至检测到第2次插入的10行数据类型不匹配
    # 此时 MTW 立刻终止所有线程，并修改状态为错误状态
    time.sleep(1)

    # 再插入1行正确数据，MTW 会因为工作线程终止而抛出异常，且不会写入该行数据
    res = writer.insert(np.datetime64('2022-03-23'),"AAAAAAAB", random.randint(1,10000))
    print("MTW has exited")
except Exception as ex:
    # MTW 抛出异常
    print("MTW exit with exception %s" % ex)
# 等待 MTW 插入完成
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 exit 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
"""

当 MTW 发生类型转换错误、写入线程写入失败等错误时，API 将会终止所有线程。此时可以通过 writer.getUnwrittenData() 方法获取失败数据，通过 insertUnwrittenData(unwrittendata) 重新写入失败数据。因为原有 MTW 对象的所有线程已经终止，无法再次被用来写入数据，因此需要重新构造一个新的 MTW 对象，将失败数据重新写入新的 MTW 对象中。

if writeStatus.hasError():
    print("Error in writing:")
    unwrittendata = writer.getUnwrittenData()
    print("Unwrittendata: %d" % len(unwrittendata))
    # 重新构造新的 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:
        # 修正失败数据后将失败数据重新写入 MTW
        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:
        # MTW 抛出异常
        print("MTW exit with exception %s" % ex)
    finally:
        # 确保 newwriter工作线程结束运行
        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
"""