关于插件:干货丨时序数据库DolphinDB插件开发教程

DolphinDB 反对动静加载内部插件，以扩大零碎性能。插件用 C ++ 编写，须要编译成 ”.so” 或 ”.dll” 共享库文件。本文着重介绍开发插件的办法和注意事项，并具体介绍以下几个具体场景的插件开发流程：

• 如何开发反对工夫序列数据处理的插件函数
• 如何开发用于解决分布式 SQL 的聚合函数
• 如何开发反对新的分布式算法的插件函数
• 如何开发反对流数据处理的插件函数
• 如何开发反对内部数据源的插件函数

1. 如何开发插件

1.1 基本概念

DolphinDB 的插件实现了能在脚本中调用的函数。一个插件函数可能是运算符函数（Operator function），也可能是零碎函数（System function），它们的区别在于，前者承受的参数个数小于等于 2，而后者的函数能够承受任意个参数，并反对会话的拜访操作。

开发一个运算符函数，须要编写一个原型为 ConstantSP (const ConstantSP& a, const ConstantSP& b) 的 C ++ 函数。当函数参数个数为 2 时，a和 b 别离为插件函数的第一和第二个参数；当参数个数为 1 时，b是一个占位符，没有理论用处；当没有参数时，a和 b 均为占位符。

开发一个零碎函数，须要编写一个原型为 ConstantSP (Heap* heap, vector<ConstantSP>& args) 的 C ++ 函数。用户在 DolphinDB 中调用插件函数时传入的参数，都按程序保留在 C ++ 的向量 args 中。heap参数不须要用户传入。

函数原型中的 ConstantSP 能够示意绝大多数 DolphinDB 对象（标量、向量、矩阵、表，等等）。其余罕用的派生自它的变量类型有VectorSP（向量）、TableSP（表）等。

1.2 创立变量

创立标量，能够间接用 new 语句创立头文件 ScalarImp.h 中申明的类型对象，并将它赋值给一个 ConstantSP。ConstantSP 是一个通过封装的智能指针，会在变量的援用计数为 0 时主动开释内存，因而，用户不须要手动 delete 曾经创立的变量：

ConstantSP i = new Int(1);                 // 相当于 1i
ConstantSP d = new Date(2019, 3, 14);      // 相当于 2019.03.14
ConstantSP s = new String("DolphinDB");    // 相当于 "DolphinDB"
ConstantSP voidConstant = new Void();      // 创立一个 void 类型变量，罕用于示意空的函数参数

头文件 Util.h 申明了一系列函数，用于疾速创立某个类型和格局的变量：

VectorSP v = Util::createVector(DT_INT, 10);     // 创立一个初始长度为 10 的 int 类型向量
v->setInt(0, 60);                                // 相当于 v[0] = 60

VectorSP t = Util::createVector(DT_ANY, 0);      // 创立一个初始长度为 0 的 any 类型向量（元组）t->append(new Int(3));                           // 相当于 t.append!(3)
t->get(0)->setInt(4);                            // 相当于 t[0] = 4
// 这里不能用 t ->setInt(0, 4)，因为 t 是一个元组，setInt(0, 4)只对 int 类型的向量无效

ConstantSP seq = Util::createIndexVector(5, 10); // 相当于 5..14
int seq0 = seq->getInt(0);                       // 相当于 seq[0]

ConstantSP mat = Util::createDoubleMatrix(5, 10);// 创立一个 10 行 5 列的 double 类型矩阵
mat->setColumn(3, seq);                          // 相当于 mat[3] = seq

1.3 异样解决和参数校验

1.3.1 异样解决

插件开发时的异样抛出和解决，和个别 C ++ 开发中一样，都通过 throw 关键字抛出异样，try语句块解决异样。DolphinDB 在头文件 Exceptions.h 中申明了异样类型。

插件函数若遇到运行时谬误，个别抛出RuntimeException。

在插件开发时，通常会校验函数参数，如果参数不符合要求，抛出一个IllegalArgumentException。罕用的参数校验函数有：

• ConstantSP->getType()：返回变量的类型（int, char, date 等等），DolphinDB 的类型定义在头文件 Types.h 中。
• ConstantSP->getCategory()：返回变量的类别，罕用的类别有 INTEGRAL（整数类型，包含 int, char, short, long 等）、FLOATING（浮点数类型，包含 float, double 等）、TEMPORAL（工夫类型，包含 time, date, datetime 等）、LITERAL（字符串类型，包含 string, symbol 等），都定义在头文件 Types.h 中。
• ConstantSP->getForm()：返回变量的格局（标量、向量、表等等），DolphinDB 的格局定义在头文件 Types.h 中。
• ConstantSP->isVector()：判断变量是否为向量。
• ConstantSP->isScalar()：判断变量是否为标量。
• ConstantSP->isTable()：判断变量是否为表。
• ConstantSP->isNumber()：判断变量是否为数字类型。
• ConstantSP->isNull()：判断变量是否为空值。
• ConstantSP->getInt()：取得变量对应的整数值，罕用于判断边界。
• ConstantSP->getString()：取得变量对应的字符串。
• ConstantSP->size()：取得变量的长度。

更多参数校验函数个别在头文件 CoreConcept.h 的Constant类办法中。

1.3.2 参数校验的范例

本节将开发一个插件函数用于求非负整数的阶乘，返回一个 long 类型变量。

DolphinDB 中 long 类型的最大值为 2^63 - 1，能示意的阶乘最大为25!，因而只有0~25 范畴内的参数是非法的。

#include "CoreConcept.h"
#include "Exceptions.h"
#include "ScalarImp.h"

ConstantSP factorial(const ConstantSP &n, const ConstantSP &placeholder) {string syntax = "Usage: factorial(n).";
    if (!n->isScalar() || n->getCategory() != INTEGRAL)
        throw IllegalArgumentException("factorial", syntax + "n must be an integral scalar.");
    int nValue = n->getInt();
    if (nValue < 0 || nValue > 25)
        throw IllegalArgumentException("factorial", syntax + "n must be a non-negative integer less than 26.");

    long long fact = 1;
    for (int i = nValue; i > 0; i--)
        fact *= i;
    return new Long(fact);
}

1.4 调用 DolphinDB 内置函数

有时会须要调用 DolphinDB 的内置函数对数据进行解决。有些类曾经定义了一些罕用的内置函数作为办法：

VectorSP v = Util::createIndexVector(1, 100);
ConstantSP avg = v->avg();     // 相当于 avg(v)
ConstantSP sum2 = v->sum2();   // 相当于 sum2(v)
v->sort(false);                // 相当于 sort(v, false)

如果须要调用其它内置函数，插件函数的类型必须是零碎函数。通过 heap->currentSession()->getFunctionDef 函数取得一个内置函数，而后用 call 办法调用它。如果该内置函数是运算符函数，应调用原型 call(Heap, const ConstantSP&, const ConstantSP&)；如果是零碎函数，应调用原型call(Heap, vector<ConstantSP>&)。以下是调用内置函数cumsum 的一个例子：

ConstantSP v = Util::createIndexVector(1, 100);
v->setTemporary(false);                                   // v 的值可能在内置函数调用时被批改。如果不心愿它被批改，应先调用 setTemporary(false)
FunctionDefSP cumsum = heap->currentSession()->getFunctionDef("cumsum");
ConstantSP result = cumsum->call(heap, v, new Void());    // 相当于 cumsum(v)，这里的 new Void()是一个占位符，没有理论用处

2. 如何开发反对工夫序列数据处理的插件函数

DolphinDB 的特色之一在于它对工夫序列有良好反对。

本章以编写一个 msum 函数的插件为例，介绍如何开发插件函数反对工夫序列数据处理。

工夫序列处理函数通常承受向量作为参数，并对向量中的每个元素进行计算解决。在本例中，msum函数承受两个参数：一个向量和一个窗口大小。它的原型是：

ConstantSP msum(const ConstantSP &X, const ConstantSP &window);

msum函数的返回值是一个和输出向量同样长度的向量。本例为简便起见，假设返回值是一个 double 类型的向量。能够通过 Util::createVector 函数事后为返回值调配空间：

int size = X->size();
int windowSize = window->getInt();
ConstantSP result = Util::createVector(DT_DOUBLE, size);

在 DolphinDB 插件编写时解决向量，能够循环应用 getDoubleConst,getIntConst 等函数，批量取得肯定长度的只读数据，保留在相应类型的缓冲区中，从缓冲区中获得数据进行计算。这样做的效率比循环应用 getDouble,getInt 等函数要高。本例为简便起见，对立应用 getDoubleConst，每次取得长度为Util::BUF_SIZE 的数据。这个函数返回一个const double*，指向缓冲区头部：

double buf[Util::BUF_SIZE];

INDEX start = 0;
while (start < size) {int len = std::min(Util::BUF_SIZE, size - start);
    const double *p = X->getDoubleConst(start, len, buf);
    for (int i = 0; i < len; i++) {double val = p[i];
        // ...
    }
    start += len;
}

在本例中，msum将计算 X 中长度为 windowSize 的窗口中所有数据的和。能够用一个长期变量 tmpSum 记录以后窗口的和，每当窗口挪动时，只有给 tmpSum 减少新窗口尾部的值，减去旧窗口头部的值，就能计算失去以后窗口中数据的和。为了将计算值写入 result，能够循环用result->getDoubleBuffer 获取一个可读写的缓冲区，写完后应用 result->setDouble 函数将缓冲区写回数组。setDouble函数会查看给定的缓冲区地址和变量底层贮存的地址是否统一，如果统一就不会产生数据拷贝。在少数状况下，用 getDoubleBuffer 取得的缓冲区就是变量理论的存储区域，这样能缩小数据拷贝，进步性能。

须要留神的是，DolphinDB 用 double 类型的最小值（曾经定义为宏 DBL_NMIN）示意double 类型的 NULL 值，要专门判断。

返回值的前 windowSize - 1 个元素为 NULL。能够对X 中的前 windowSize 个元素和之后的元素用两个循环别离解决，前一个循环只计算累加，后一个循环执行加和减的操作。最终的实现如下：

ConstantSP msum(const ConstantSP &X, const ConstantSP &window) {INDEX size = X->size();
    int windowSize = window->getInt();
    ConstantSP result = Util::createVector(DT_DOUBLE, size);

    double buf[Util::BUF_SIZE];
    double windowHeadBuf[Util::BUF_SIZE];
    double resultBuf[Util::BUF_SIZE];
    double tmpSum = 0.0;

    INDEX start = 0;
    while (start < windowSize) {int len = std::min(Util::BUF_SIZE, windowSize - start);
        const double *p = X->getDoubleConst(start, len, buf);
        double *r = result->getDoubleBuffer(start, len, resultBuf);
        for (int i = 0; i < len; i++) {if (p[i] != DBL_NMIN)    // p[i] is not NULL
                tmpSum += p[i];
            r[i] = DBL_NMIN;
        }
        result->setDouble(start, len, r);
        start += len;
    }

    result->setDouble(windowSize - 1, tmpSum);    // 上一个循环多设置了一个 NULL，填充为 tmpSum

    while (start < size) {int len = std::min(Util::BUF_SIZE, size - start);
        const double *p = X->getDoubleConst(start, len, buf);
        const double *q = X->getDoubleConst(start - windowSize, len, windowHeadBuf);
        double *r = result->getDoubleBuffer(start, len, resultBuf);
        for (int i = 0; i < len; i++) {if (p[i] != DBL_NMIN)
                tmpSum += p[i];
            if (q[i] != DBL_NMIN)
                tmpSum -= q[i];
            r[i] = tmpSum;
        }
        result->setDouble(start, len, r);
        start += len;
    }

    return result;
}

3. 如何开发用于解决分布式 SQL 的聚合函数

在 DolphinDB 中，SQL 的聚合函数通常承受一个或多个向量作为参数，最终返回一个标量。在开发聚合函数的插件时，须要理解如何拜访向量中的元素。

DolphinDB 中的向量有两种存储形式。一种是惯例数组，数据在内存中间断存储；另一种是大数组，其中的数据分块存储。

本章将以编写一个求几何平均数的函数为例，介绍如何开发聚合函数，重点关注数组中元素的拜访。

3.1 聚合函数范例

几何平均数 geometricMean 函数承受一个向量作为参数。为了避免溢出，个别采纳其对数模式计算，即

geometricMean([x1, x2, ..., xn])
    = exp((log(x1) + log(x2) + log(x3) + ... + log(xn))/n)

为了实现这个函数的分布式版本，能够先开发聚合函数插件 logSum，用以计算某个分区上的数据的对数和，而后用defg 关键字定义一个 Reduce 函数，用 mapr 关键字定义一个 MapReduce 函数。

在 DolphinDB 插件开发中，对数组的操作通常要思考它是惯例数组还是大数组。能够用 isFastMode 函数判断：

ConstantSP logSum(const ConstantSP &x, const ConstantSP &placeholder) {if (((VectorSP) x)->isFastMode()) {// ...}
    else {// ...}
}

如果数组是惯例数组，它在内存中间断存储。能够用 getDataArray 函数取得它数据的指针。假设数据是以 double 类型存储的：

if (((VectorSP) x)->isFastMode()) {int size = x->size();
    double *data = (double *) x->getDataArray();
    
    double logSum = 0;
    for (int i = 0; i < size; i++) {if (data[i] != DBL_NMIN)    // is not NULL
            logSum += std::log(data[i]);
    }
    return new Double(logSum);
}

如果数据是大数组，它在内存中分块存储。能够用 getSegmentSize 取得每个块的大小，用 getDataSegment 取得首个块的地址。它返回一个二级指针，指向一个指针数组，这个数组中的每个元素指向每个块的数据数组：

// ...
else {int size = x->size();
    int segmentSize = x->getSegmentSize();
    double **segments = (double **) x->getDataSegment();
    INDEX start = 0;
    int segmentId = 0;
    double logSum = 0;

    while (start < size) {double *block = segments[segmentId];
        int blockSize = std::min(segmentSize, size - start);
        for (int i = 0; i < blockSize; i++) {if (block[i] != DBL_NMIN)    // is not NULL
                logSum += std::log(block[i]);
        }
        start += blockSize;
        segmentId++;
    }
    return new Double(logSum);
}

在理论开发中，数组的底层存储不肯定是 double 类型。用户须要思考具体类型。本例采纳了泛型编程对立解决不同类型，具体代码参见附件。

3.2 在 DolphinDB 中调用函数

通常须要实现一个聚合函数的非分布式版本和分布式版本，零碎会基于哪个版本更高效来抉择调用这个版本。

在 DolphinDB 中定义非分布式的 geometricMean 函数：

def geometricMean(x) {return exp(logSum::logSum(x)  count(x))
}

而后通过定义 Map 和 Reduce 函数，最终用 mapr 定义分布式的版本：

def geometricMeanMap(x) {return logSum::logSum(x)
}

defg geometricMeanReduce(myLogSum, myCount) {return exp(sum(myLogSum)  sum(myCount))
}

mapr geometricMean(x) {geometricMeanMap(x), count(x) -> geometricMeanReduce }

这样就实现了 geometricMean 函数。

如果是在单机环境中执行这个函数，只须要在执行的节点上加载插件。如果有数据位于近程节点，须要在每一个近程节点加载插件。能够手动在每个节点执行 loadPlugin 函数，也能够用以下脚本疾速在每个节点上加载插件：

each(rpc{, loadPlugin, pathToPlugin}, getDataNodes())

通过以下脚本创立一个分区表，验证函数：

db = database("", VALUE, 1 2 3 4)
t = table(take(1..4, 100) as id, rand(1.0, 100) as val)
t0 = db.createPartitionedTable(t, `tb, `id)
t0.append!(t)
select geometricMean(val) from t0 group by id

3.3 随机拜访大数组

能够对大数组进行随机拜访，但要通过下标计算。用 getSegmentSizeInBit 函数取得块大小的二进制位数，通过位运算取得块的偏移量和块内偏移量：

int segmentSizeInBit = x->getSegmentSizeInBit();
int segmentMask = (1 << segmentSizeInBit) - 1;
double **segments = (double **) x->getDataSegment();

int index = 3000000;    // 想要拜访的下标

double result = segments[index >> segmentSizeInBit][index & segmentMask];
//                       ^ 块的偏移量                ^ 块内偏移量

3.4 应该抉择哪种形式拜访向量

上一章介绍了通过 getDoubleConst,getIntConst 等一族办法取得只读缓冲区，以及通过 getDoubleBuffer,getIntBuffer 等一族办法取得可读写缓冲区，这两种拜访向量的办法。本章介绍了通过 getDataArray 和getDataSegment办法间接拜访向量的底层存储。在理论开发中，前一种办法更通用，个别应该抉择前一种办法。但在某些特别的场合（例如明确晓得数据存储在大数组中，且晓得数据的类型），能够采纳第二种办法。

4. 如何开发反对新的分布式算法的插件函数

在 DolphinDB 中，Map-Reduce 是执行分布式算法的通用计算框架。DolphinDB 提供了 mr 函数和 imr 函数，使用户能通过脚本实现分布式算法。而在编写分布式算法的插件时，应用的同样是这两个函数。本章次要介绍如何用 C ++ 语言编写自定义的 map, reduce 等函数，并调用 mr 和 imr 两个函数，最终实现分布式计算。

4.1 分布式算法范例

本章将以 mr 为例，实现一个函数，求分布式表中相应列名的所有列平均值，介绍编写 DolphinDB 分布式算法插件的整体流程，及须要留神的技术细节。

在插件开发中，用户自定义的 map, reduce, final, term 函数，能够是运算符函数，也能够是零碎函数。

本例的 map 函数，对表的一个分区内对应列名的列做计算，返回一个长度为 2 的元组，别离蕴含数据的和，及数据非空元素的个数。具体实现如下：

ConstantSP columnAvgMap(Heap *heap, vector<ConstantSP> &args) {TableSP table = args[0];
    ConstantSP colNames = args[1];
    double sum = 0.0;
    int count = 0;
    
    for (int i = 0; i < colNames->size(); i++) {string colName = colNames->getString(i);
        VectorSP col = table->getColumn(colName);
        sum += col->sum()->getDouble();
        count += col->count();}

    ConstantSP result = Util::createVector(DT_ANY, 2);
    result->set(0, new Double(sum));
    result->set(1, new Int(count));
    return result;
}

本例的 reduce 函数，是对 map 后果的相加。DolphinDB 的内置函数 add 就提供了这个性能，能够用 heap->currentSession()->getFunctionDef("add") 取得这个函数：

FunctionDefSP reduceFunc = heap->currentSession()->getFunctionDef("add");

本例的 final 函数，是对 reduce 后果中的数据总和 sum 和非空元素个数 count 做除法，求得所有分区中对应列的平均数。具体实现如下：

ConstantSP columnAvgFinal(const ConstantSP &result, const ConstantSP &placeholder) {double sum = result->get(0)->getDouble();
    int count = result->get(1)->getInt();
    
    return new Double(sum / count);
}

定义了 map, reduce, final 等函数后，将它们导出为插件函数（在头文件的函数申明前加上 extern "C"，并在加载插件的文本文件中列出这些函数），而后通过heap->currentSession->getFunctionDef 获取这些函数，就能以这些函数为参数调用 mr 函数。如：

FunctionDefSP mapFunc = Heap->currentSession()->getFunctionDef("columnAvg::columnAvgMap");

在本例中，map 函数承受两个参数 table 和colNames，但 mr 只容许 map 函数有一个参数，因而须要以局部利用的模式调用 map 函数，能够用 Util::createPartialFunction 将它包装为局部利用，实现如下：

vector<ConstantSP> mapWithColNamesArgs {new Void(), colNames};
FunctionDefSP mapWithColNames = Util::createPartitalFunction(mapFunc, mapWithColNamesArgs);

用 heap->currentSession()->getFunctionDef("mr") 取得零碎内置函数 mr，调用mr->call 办法，就相当于在 DolphinDB 脚本中调用 mr 函数。最初实现的 columnAvg 函数定义如下：

ConstantSP columnAvg(Heap *heap, vector<ConstantSP> &args) {ConstantSP ds = args[0];
    ConstantSP colNames = args[1];

    FunctionDefSP mapFunc = heap->currentSession()->getFunctionDef("columnAvg::columnAvgMap");
    vector<ConstantSP> mapWithColNamesArgs = {new Void(), colNames};
    FunctionDefSP mapWithColNames = Util::createPartialFunction(mapFunc, mapWithColNamesArgs);    // columnAvgMap{, colNames}
    FunctionDefSP reduceFunc = heap->currentSession()->getFunctionDef("add");
    FunctionDefSP finalFunc = heap->currentSession()->getFunctionDef("columnAvg::columnAvgFinal");

    FunctionDefSP mr = heap->currentSession()->getFunctionDef("mr");    // mr(ds, columnAvgMap{, colNames}, add, columnAvgFinal)
    vector<ConstantSP> mrArgs = {ds, mapWithColNames, reduceFunc, finalFunc};
    return mr->call(heap, mrArgs);
}

4.2 在 DolphinDB 中调用函数

如果是在单机环境中执行这个函数，只须要在执行的节点上加载插件。但如果有数据位于近程节点，须要在每一个近程节点加载插件。能够手动在每个节点执行 loadPlugin 函数，也能够用以下脚本疾速在每个节点上加载插件：

each(rpc{, loadPlugin, pathToPlugin}, getDataNodes())

加载插件后，用 sqlDS 函数生成数据源，并调用函数：

n = 100
db = database("dfs://testColumnAvg", VALUE, 1..4)
t = db.createPartitionedTable(table(10:0, `id`v1`v2, [INT,DOUBLE,DOUBLE]), `t, `id)
t.append!(table(take(1..4, n) as id, rand(10.0, n) as v1, rand(100.0, n) as v2))

ds = sqlDS(<select * from t>)
columnAvg::columnAvg(ds, `v1`v2)

5. 如何开发反对流数据处理的插件函数

在 DolphinDB 中，流数据订阅端能够通过一个 handler 函数解决收到的数据。订阅数据能够是一个数据表，或一个元组，由 subsrciebeTable 函数的 msgAsTable 参数决定。通常能够用 handler 函数对流数据进行过滤、插入另一张表等操作。

本章将编写一个 handler 函数。它承受的音讯类型是元组。另外承受两个参数：一个是 int 类型的标量或向量indices，示意元组中元素的下标，另一个是一个表table。它将元组中对应下标的列插入到表中。

向表中增加数据的接口是 bool append(vector<ConstantSP>& values, INDEX& insertedRows, string& errMsg)，如果插入胜利，返回true，并向insertedRows 中写入插入的行数。否则返回 false，并在errMsg 中写入出错信息。插件的实现如下：

ConstantSP handler(Heap *heap, vector<ConstantSP> &args) {ConstantSP indices = args[0];
    TableSP table = args[1];
    ConstantSP msg = args[2];

    vector<ConstantSP> msgToAppend;
    for (int i = 0; i < indices->size(); i++) {int index = indices->get(i);
        msgToAppend.push_back(msg->get(index));
    }

    INDEX insertedRows;
    string errMsg;
    table->append(msgToAppend, insertedRows, errMsg);
    return new Void();}

在理论利用中，可能须要晓得插入出错时的起因。能够引入头文件Logger.h，将出错信息写入日志中。留神须要在编译插件时加上宏定义-DLOGGING_LEVEL_2：

// ...
bool success = table->append(msgToAppend, insertedRows, errMsg);
if (!success)
    LOG_ERR("Failed to append to table:", errMsg);

能够用以下脚本模仿流数据写入，验证 handler 函数：

loadPlugin("/path/to/PluginHandler.txt")

share streamTable(10:0, `id`sym`timestamp, [INT,SYMBOL,TIMESTAMP]) as t0
t1 = table(10:0, `sym`timestamp, [SYMBOL,TIMESTAMP])
subscribeTable(, `t0, , , handler::handler{[1,2], t1})

t0.append!(table(1..100 as id, take(`a`b`c`d, 100) as symbol, now() + 1..100 as timestamp))

select * from t1

6. 如何开发反对内部数据源的插件函数

在为第三方数据设计可扩大的接口插件时，有几个须要关注的问题：

1. 数据源（Data source）。数据源是一个非凡的数据对象，蕴含了数据实体的元形容，执行一个数据源能取得数据实体，可能是表、矩阵、向量等等。用户能够提供数据源调用 olsEx, randomForestClassifier 等分布式计算函数，也能够调用 mr, imr 或ComputingModel.h中定义的更底层的计算模型做并行计算。DolphinDB 的内置函数 sqlDS 就通过 SQL 表达式获取数据源。在设计第三方数据接口时，通常须要实现一个获取数据源的函数，它将大的文件分成若干个局部，每局部都示意数据的一个子集，最初返回一个数据源的元组。数据源个别用一个 Code object 示意，是一个函数调用，它的参数是元数据，返回一个表。
2. 构造（Schema）。表的构造形容了表的列数，每一列的列名和数据类型。第三方接口通常须要实现一个函数，疾速取得数据的表构造，以便用户在这个构造的根底上调整列名和列的数据类型。
3. IO 问题。在多核多 CPU 的环境中，IO 可能成为瓶颈。DolphinDB 提供了形象的 IO 接口，DataInputStream和 DataOutputStream，这些接口封装了数据压缩，Endianness，IO 类型（网络，磁盘，buffer 等）等细节，不便开发。此外还特地实现了针对多线程的 IO 实现，BlockFileInputStream 和BlockFileOutputStream。这个实现有两个长处：

• 实现计算和 IO 并行。A 线程在解决数据的时候，后盾线程在异步帮 A 线程预读取前面须要的数据。
• 防止了多线程的磁盘竞争。当线程个数减少的时候，如果并行往同一个磁盘上读写，性能会急剧下降。这个实现，会对同一个磁盘的读写串行化，从而进步吞吐量。

本章将介绍通常须要实现的几个函数，为设计第三方数据接口提供一个简略的范例。

6.1 数据格式形容

假设本例中的数据贮存在立体文件数据库，以二进制格局按行存储，数据从文件头部间接开始存储。每行有四列，别离为 id（按有符号 64 位长整型格局存储，8 字节），symbol（按 C 字符串格局存储，8 字节），date（按 BCD 码格局存储，8 字节），value（按 IEEE 754 规范的双精度浮点数格局存储，8 字节），每行共 32 字节。以下是一行的例子：

这一行的十六进制示意为：

0x 00 00 00 00 00 00 00 05
0x 49 42 4D 00 00 00 00 00
0x 02 00 01 09 00 03 01 03
0x 40 24 33 33 33 33 33 33

6.2 extractMyDataSchema 函数

这个函数提取数据文件的表构造。在本例中，表构造是确定的，不须要理论读取文件。该函数提供了一个如何生成表构造的范例。它通过 Util::createTable 函数创立一张构造表：

ConstantSP extractMyDataSchema(const ConstantSP &placeholderA, const ConstantSP &placeholderB) {ConstantSP colNames = Util::createVector(DT_STRING, 4);
    ConstantSP colTypes = Util::createVector(DT_STRING, 4);
    string names[] = {"id", "symbol", "date", "value"};
    string types[] = {"LONG", "SYMBOL", "DATE", "DOUBLE"};
    colNames->setString(0, 4, names);
    colTypes->setString(0, 4, types);

    vector<ConstantSP> schema = {colNames, colTypes};
    vector<string> header = {"name", "type"};

    return Util::createTable(header, schema);
}

在理论开发中，可能须要以读取文件头等形式取得表构造。如何读文件将在前面介绍。

6.3 loadMyData 函数

loadMyData函数读取文件，并输入一张 DolphinDB 表。给定一个文件的门路，能够通过 Util::createBlockFileInputStream 创立一个输出流，尔后，可对这个流调用 readBytes 函数读取给定长度的字节，readBool读取下一个 bool 值，readInt读取下一个 int 值，等等。本例给 loadMyData 函数设计的语法为：loadMyData(path, [start], [length])。除了承受文件门路 path，还承受两个int 类型的参数 start 和length，别离示意开始读取的行数和须要读取的总行数。createBlockFileInputStream函数能够通过参数决定开始读取的字节数和须要读取的总字节数：

ConstantSP loadMyData(Heap *heap, vector<ConstantSP> &args) {ConstantSP path = args[0];
    long long fileLength = Util::getFileLength(path->getString());
    size_t bytesPerRow = 32;

    int start = args.size() >= 2 ? args[1]->getInt() : 0;
    int length = args.size() >= 3 ? args[2]->getInt() : fileLength / bytesPerRow - start;

    DataInputStreamSP inputStream = Util::createBlockFileInputStream(path->getString(), 0, fileLength, Util::BUF_SIZE, start * bytesPerRow, length * bytesPerRow);
    char buf[Util::BUF_SIZE];
    size_t actualLength;
    
    while (true) {inputStream->readBytes(buf, Util::BUF_SIZE, actualLength);
        if (actualLength <= 0)
            break;
        // ...
    }
}

在读取数据时，通常将数据缓存到数组中，期待缓冲区满后批量插入。例如，假设要读取一个内容全为 char 类型字节的二进制文件，将它写入一个 char 类型的 DolphinDB 向量 vec。最初返回只由vec 一列组成的表：

char buf[Util::BUF_SIZE];
VectorSP vec = Util::createVector(DT_CHAR, 0);
size_t actualLength;

while (true) {inputStream->readBytes(buf, Util::BUF_SIZE, actualLength);
    if (actualLength <= 0)
        break;
    vec->appendChar(buf, actualLength);
}

vector<ConstantSP> cols = {vec};
vector<string> colNames = {"col0"};

return Util::createTable(colNames, cols);

本节的残缺代码请参考附件中的代码。在理论开发中，加载数据的函数可能还会承受表构造参数schema，按理论须要扭转读取的数据类型。

6.4 loadMyDataEx函数

loadMyData函数总是将数据加载到内存，当数据文件十分宏大时，工作机的内存很容易成为瓶颈。所以设计 loadMyDataEx 函数解决这个问题。它通过边导入边保留的形式，把动态的二进制文件以较为平缓的数据流的形式保留为 DolphinDB 的分布式表，而不是采纳全副导入内存再存为分区表的形式，从而升高内存的应用需要。

loadMyDataEx函数的参数能够参考 DolphinDB 内置函数 loadTextEx。它的语法是：loadMyDataEx(dbHandle, tableName, partitionColumns, path, [start], [length])。如果数据库中的表存在，则将导入的数据增加到已有的表result 中。如果表不存在，则创立一张表result，而后增加数据。最初返回这张表：

string dbPath = ((SystemHandleSP) db)->getDatabaseDir();
vector<ConstantSP> existsTableArgs = {new String(dbPath), tableName};
bool existsTable = heap->currentSession()->getFunctionDef("existsTable")->call(heap, existsTableArgs)->getBool();    // 相当于 existsTable(dbPath, tableName)
ConstantSP result;

if (existsTable) {    // 表存在，间接加载表
    vector<ConstantSP> loadTableArgs = {db, tableName};
    result = heap->currentSession()->getFunctionDef("loadTable")->call(heap, loadTableArgs);    // 相当于 loadTable(db, tableName)
}
else {    // 表不存在，创立表
    TableSP schema = extractMyDataSchema(new Void(), new Void());
    ConstantSP dummyTable = DBFileIO::createEmptyTableFromSchema(schema);
    vector<ConstantSP> createTableArgs = {db, dummyTable, tableName, partitionColumns};
    result = heap->currentSession()->getFunctionDef("createPartitionedTable")->call(heap, createTableArgs);    // 相当于 createPartitionedTable(db, dummyTable, tableName, partitionColumns)
}

读取数据并增加到表中的代码实现采纳了 Pipeline 框架。它的初始工作是一系列具备不同 start 参数的 loadMyData 函数调用，pipeline 的 follower 函数是一个局部利用 append!{result}，相当于把整个读取数据的工作分成若干份执行，调用loadMyData 分块读取后，将相应的数据通过 append! 插入表中。外围局部的代码如下：

int sizePerPartition = 16 * 1024 * 1024;
int partitionNum = fileLength / sizePerPartition;
vector<DistributedCallSP> tasks;
FunctionDefSP func = Util::createSystemFunction("loadMyData", loadMyData, 1, 3, false);
int partitionStart = start;
int partitionLength = length / partitionNum;
for (int i = 0; i < partitionNum; i++) {if (i == partitionNum - 1)
        partitionLength = length - partitionLength * i;
    vector<ConstantSP> partitionArgs = {path, new Int(partitionStart), new Int(partitionLength)};
    ObjectSP call = Util::createRegularFunctionCall(func, partitionArgs);    // 将会调用 loadMyData(path, partitionStart, partitionLength)
    tasks.push_back(new DistributedCall(call, true));
    partitionStart += partitionLength;
}

vector<ConstantSP> appendToResultArgs = {result};
FunctionDefSP appendToResult = Util::createPartialFunction(heap->currentSession()->getFunctionDef("append!"), appendToResultArgs);    // 相当于 append!{result}
vector<FunctionDefSP> functors = {appendToResult};
PipelineStageExecutor executor(functors, false);
executor.execute(heap, tasks);

本节的残缺代码请参考附件中的代码。用 Pipeline 框架实现数据的分块导入，只是一种思路。在具体开发时，能够采纳 ComputingModel.h 中申明的 StaticStageExecutor，也能够应用Concurrent.h 中申明的线程模型Thread。实现办法有很多种，须要依据理论场景抉择。

6.5 myDataDS函数

myDataDS函数返回一个数据源的元组。每个数据源都是一个示意函数调用的 Code object，能够通过 Util::createRegularFunctionCall 生成。执行这个对象能够获得对应的数据。以下是基于 loadMyData 函数产生数据源的一个例子：

ConstantSP myDataDS(Heap *heap, vector<ConstantSP> &args) {ConstantSP path = args[0];
    long long fileLength = Util::getFileLength(path->getString());
    size_t bytesPerRow = 32;

    int start = args.size() >= 2 ? args[1]->getInt() : 0;
    int length = args.size() >= 3 ? args[2]->getInt() : fileLength / bytesPerRow - start;

    int sizePerPartition = 16 * 1024 * 1024;
    int partitionNum = fileLength / sizePerPartition;

    int partitionStart = start;
    int partitionLength = length / partitionNum;

    FunctionDefSP func = Util::createSystemFunction("loadMyData", loadMyData, 1, 3, false);
    ConstantSP dataSources = Util::createVector(DT_ANY, partitionNum);
    for (int i = 0; i < partitionNum; i++) {if (i == partitionNum - 1)
            partitionLength = length - partitionLength * i;
        vector<ConstantSP> partitionArgs = {path, new Int(partitionStart), new Int(partitionLength)};
        ObjectSP code = Util::createRegularFunctionCall(func, partitionArgs);    // 将会调用 loadMyData(path, partitionStart, partitionLength)
        dataSources->set(i, new DataSource(code));
    }
    return dataSources;
}

教程中的残缺代码见 https://github.com/dolphindb/Tu