mysql

支持 DolphinDB Server 2.00.10 及更高版本；支持 Shark；支持 Linux x86-64，Linux ABI，Windows x64。

1. 在DolphinDB 客户端中使用 listRemotePlugins 命令查看插件仓库中的插件信息。

   注意：仅展示当前操作系统和 注意：仅展示当前操作系统和 server 版本支持的插件。若无预期插件，可自行编译或在 DolphinDB 用户社区进行反馈。

   login("admin", "123456")
   listRemotePlugins()

2. 使用 installPlugin 命令完成插件安装。

   installPlugin("mysql")

3. 使用 loadPlugin 命令加载插件。

   loadPlugin("mysql")

语法

mysql::connect(host, port, user, password, db, [config])

详情

与 MySQL 服务器建立一个连接。返回一个 MySQL 连接的句柄，用于 load 与 loadEx 等操作。

参数

host MySQL 服务器的地址，类型为 string。

port MySQL 服务器的端口，类型为 int。

user MySQL 服务器的用户名，类型为 string。

password MySQL 服务器的密码，类型为 string。

db 要使用的数据库名称，类型为 string。

config 可选参数，指定连接配置项的字典。字典的 key 是 STRING 类型，代表配置名称，value 是配置内容。

• "SSL_ENFORCE" 指定是否开启 SSL 功能，类型为 bool，默认是 true。
• "SSL_VERIFY_SERVER_CERT" 指定是否验证证书，类型为 bool，默认为 false。
• "CHARSET" 指定连接使用的字符集，类型为 string，默认为"UTF8"。

示例

conn = mysql::connect(`127.0.0.1, 3306, `root, `root, `DolphinDB)

语法

mysql::showTables(conn)

详情

列出建立 MySQL 连接时指定的数据库中包含的所有表。

参数

conn 通过 mysql::connect 获得的 MySQL 连接句柄。

示例

conn = mysql::connect(`192.168.1.16, 3306, `root, `root, `DolphinDB)
mysql::showTables(conn)

output:
  Tables_in_DolphinDB
  -------------------
  US

语法

mysql::extractSchema(conn, tableName)

详情

生成指定数据表的结构。

参数

conn 通过 mysql::connect 获得的 MySQL 连接句柄。

tableName MySQL 表名，类型为 string。

示例

conn = mysql::connect(`192.168.1.16, 3306, `root, `root, `DolphinDB)
mysql::extractSchema(conn, `US)

output:
        name    type   DolphinDBType
        PERMNO  int(11)     INT
        date    date        DATE
        SHRCD   int(11)     INT
        TICKER  varchar(10) STRING
        ...
        PRC     double      DOUBLE

语法

mysql::load(conn, table|query, [schema], [startRow], [rowNum], [allowEmptyTable])

详情

将 MySQL 表或者 SQL 查询结果导入 DolphinDB 中的内存表。支持的数据类型以及数据转化规则可见用户手册数据类型章节。

参数

conn 通过 mysql::connect 获得的 MySQL 连接句柄。

table|query 一张 MySQL 中表的名字，或者类似 select * from table limit 100 的合法 MySQL 查询语句，类型为 string。

schema 是一个包含列名和列类型信息的表，两列均为 string 类型，第一列是列名，用于创建结果表，第二列是要转换成的列类型。该表可以有更多列，只要前两列满足要求。如果想要改变由系统自动决定的列的数据类型，需要在 schema 表中修改数据类型，并且把它作为 load 函数的一个参数。除了手动创建，该参数还可以通过 extractSchema 接口来获取。

startRow 读取 MySQL 表的起始行数，若不指定，默认从数据集起始位置读取。若 'table|query' 是查询语句，则这个参数不起作用。

rowNum 读取 MySQL 表的行数，若不指定，默认读到数据集的结尾。若 'table|query' 是查询语句，则这个参数不起作用。

allowEmptyTable 一个布尔值，表示是否允许从 MySQL 读取空表（没有数据），默认为不允许。该配置项用于管控对空表的加载限制。

示例

通过表名加载数据，并指定起始行和加载的行数。

conn = mysql::connect(`192.168.1.18, 3306, `root, `root, `DolphinDB)
tb = mysql::load(conn,`US,,1000,10000)
select count(*) from tb

通过 SQL 语句加载数据。

conn = mysql::connect(`127.0.0.1, 3306, `root, `root, `DolphinDB)
tb = mysql::load(conn, "SELECT PERMNO FROM US LIMIT 123456")
select count(*) from tb

如下是一个通过 extractSchema 接口获取 schema 参数，修改后调用 load 接口加载数据到内存表的示例：

// 通过 extractSchema 获取 example 表的结构，并用 update 将其中一列的类型改成 double
schema = mysql::extractSchema(conn, "example")
update schema set type = "double" where name = "column"

// 加载 example 表，此时 column 列的类型已被转换成 double
tb = mysql::load(conn, "example", schema)

语法

mysql::loadEx(conn, dbHandle,tableName,partitionColumns,table|query,[schema],[startRow],[rowNum],[transform],[sortColumns],[keepDuplicates],[sortKeyMappingFunction])

详情

将 MySQL 中的表或查询结果转换为 DolphinDB 数据库的分布式表，然后将表的元数据加载到内存中。支持的数据类型，以及数据转化规则可见数据类型章节。

参数

conn 通过 mysql::connect 获得的 MySQL 连接句柄。

dbHandle 与 tableName 若要将输入数据文件保存在分布式数据库中，需要指定数据库句柄和表名。

partitionColumns 字符串标量或向量，表示分区列。在组合分区中，partitionColumns是字符串向量。

table|query MySQL 中表的名字，或者类似 select * from table limit 100 的合法 MySQL 查询语句，类型为 string。需要注意查询的 MySQL 表的列顺序需要与 DolphinDB 分布式表的列顺序保持一致，否则会出现值错误或类型转换失败的问题。

schema 是一个包含列名和列类型信息的表，两列均为 string 类型，第一列是列名，用于创建结果表，如果分布式表已存在则无作用，第二列是要转换成的列类型。该表可以有更多列，只要前两列满足要求。若要修改由系统自动检测的列的数据类型，需要在 schema 表中修改数据类型，并且把它作为 load 函数的一个参数。除了手动创建，该参数还可以通过 extractSchema 接口来获取。

startRow 读取 MySQL 表的起始行数，若不指定，默认从数据集起始位置读取。若 'table|query' 是查询语句，则这个参数不起作用。

rowNum 读取 MySQL 表的行数，若不指定，默认读到数据集的结尾。若 'table|query' 是查询语句，则这个参数不起作用。

transform 导入到 DolphinDB 数据库前对 MySQL 表进行转换，例如替换列。

sortColumns 字符串标量或向量，用于指定表的排序列，写入的数据将按 sortColumns 进行排序，只在创建 TSDB 引擎表时需要。

keepDuplicates 指定在每个分区内如何处理所有 sortColumns 之值皆相同的数据，ALL 用于保留所有数据，为默认值，LAST 仅保留最新数据，FIRST 仅保留第一条数据，只在创建 TSDB 引擎表时需要。

sortKeyMappingFunction 由一元函数对象组成的向量，其长度与索引列一致，即 sortColumns 的长度 - 1。用于指定应用在索引列中各列的映射函数，以减少 sort key 的组合数，该过程称为 sort key 降维，只在创建 TSDB 引擎表时需要。

示例

• 将数据导入磁盘上的分区表

  dbPath = "C:/..."
  db = database(dbPath, RANGE, 0 500 1000)
  mysql::loadEx(conn, db,`tb, `PERMNO, `US)
  tb = loadTable(dbPath, `tb)

• 将数据导入内存中的分区表

  直接原表导入

  db = database("", RANGE, 0 50000 10000)
  tb = mysql::loadEx(conn, db,`tb, `PERMNO, `US)

通过SQL导入

db = database("", RANGE, 0 50000 10000)
tb = mysql::loadEx(conn, db,`tb, `PERMNO, "SELECT * FROM US LIMIT 100");

• 将数据导入DFS分布式文件系统中的分区表

直接原表导入

db = database("dfs://US", RANGE, 0 50000 10000)
mysql::loadEx(conn, db,`tb, `PERMNO, `US)
tb = loadTable("dfs://US", `tb)

通过SQL导入

db = database("dfs://US", RANGE, 0 50000 10000)
mysql::loadEx(conn, db,`tb, `PERMNO, "SELECT * FROM US LIMIT 1000");
tb = loadTable("dfs://US", `tb)

导入前对 MySQL 表进行转换

db = database("dfs://US", RANGE, 0 50000 10000)
def replaceTable(mutable t){
	return t.replaceColumn!(`svalue,t[`savlue]-1)
}
t=mysql::loadEx(conn, db, "",`stockid, 'select  * from US where stockid<=1000000',,,,replaceTable)

获取 schema 参数并加载数据到内存表

// 通过 extractSchema 获取 example 表的结构，并用 update 将其中一列的类型改成 double
schema = mysql::extractSchema(conn, "example")
update schema set type = "double" where name = "column"

// 创建一个数据库，然后 loadEx 加载 example 表并保存到 dfs://example 数据库的 pt 表中
db = database("dfs://example", RANGE, 0 50000 1000000 1500000 2000000 2500000 3000001)
mysql::loadEx(conn, db, `pt, `partitionColumn, "example", schema)

语法

mysql::subscribeBinlog(conn, sourceTables, targetTable, [options])

详情

创建 Binlog 订阅，实时捕获指定 MySQL 表的 INSERT、UPDATE 和 DELETE 操作，并将 MySQL 变化数据捕获（Change Data Capture， CDC）记录异步追加到 DolphinDB 表。存在有效 checkpoint 时从 checkpoint 恢复，否则从调用时 MySQL 的当前 Binlog 位点开始订阅。

使用该接口前，需通过 mysql::connect 建立连接。MySQL 服务必须设置 log_bin=ON、binlog_format=ROW 和 binlog_row_image=FULL。连接用户需要具备读取 Binlog、查询当前 Binlog 位点和访问源表结构的权限。源表不能包含 MySQL 原生 JSON 类型列。

订阅建立成功后，传入的 conn 可以继续用于其他查询，也可以单独关闭，但不得让 mysql::close(conn) 与 mysql::subscribeBinlog(conn, ...) 并发执行。订阅期间源表结构发生变化时，应停止并重新创建订阅。

参数

conn 通过 mysql::connect 创建且处于打开状态的 MySQL 连接句柄。接口使用该连接的配置创建独立的元数据连接和复制连接。

sourceTables 包含订阅源表信息的表。该表不能为空，且同一源表不能重复。必须包含以下两列：

targetTable 接收 CDC 数据的 DolphinDB 表。必须严格包含以下 9 列，列名、顺序和类型均不可改变：

不同操作对应的行数据如下：

超过 JSON 安全整数范围的整数和 DECIMAL 在 before、after 中以 JSON 字符串表示；二进制、BIT 和 GEOMETRY 数据以 0x... 表示。

options 可选参数，一个 (STRING/SYMBOL -> ANY) 类型的字典。支持以下配置项：

不支持通过 binlogFile 或 binlogPosition 手工指定起始位点。

checkpointTable 必须以 subscriptionName 作为首个分区列，并严格包含以下 7 列：

恢复订阅时，必须使用相同的 subscriptionName、源表和目标表。插件根据 checkpointSeq 最大的记录恢复，并使用 (binlogFile, eventPos, rowIndex) 避免重复写入已存在的 CDC 记录。

返回值

返回 MySQL Binlog 订阅资源。应保存该资源；资源被释放时订阅会自动停止。

示例

conn = mysql::connect("127.0.0.1", 3306, "test_user", "123456", "test2")

sourceTables = table(
    ["test2"] as databaseName,
    ["binlog_case"] as tableName
)

targetTable = table(
    1000:0,
    `databaseName`tableName`operation`eventTime`binlogFile`eventPos`rowIndex`before`after,
    [SYMBOL, SYMBOL, SYMBOL, TIMESTAMP, STRING, LONG, INT, STRING, STRING]
)

options = dict(STRING, ANY)
options["subscriptionName"] = "mysql_binlog_case"

subscription = mysql::subscribeBinlog(conn, sourceTables, targetTable, options)

// 订阅成功后，在 MySQL 中对 test2.binlog_case 执行 INSERT、UPDATE 或 DELETE。
select * from targetTable

语法

mysql::unsubscribeBinlog(subscription)

详情

停止指定订阅，等待后台线程退出并释放订阅使用的连接。停止后不会继续向 CDC 表或 checkpoint 表写入数据，也不会关闭创建订阅时传入的 conn。状态为 FAILED 的订阅也应调用该接口释放资源。

参数

subscription 可以是 subscribeBinlog 返回的订阅资源，也可以是非空的订阅名称 STRING 标量。通过名称停止订阅时，调用者必须是订阅所有者或管理员。

返回值

无

示例

mysql::unsubscribeBinlog(subscription)

// 也可以通过名称停止。
mysql::unsubscribeBinlog("mysql_binlog_case")

语法

mysql::getBinlogStatus([subscription])

详情

查询 Binlog 订阅的状态和处理进度。指定订阅资源时返回该订阅的一行状态；不指定参数时返回当前用户可见且尚未释放的订阅，管理员可以查看全部此类订阅。订阅停止后，只要资源仍未释放，仍可通过资源查询最终状态。

参数

subscription 可选参数，subscribeBinlog 返回的订阅资源。

返回值

状态表，包含以下列：

示例

// 查询指定订阅。
status = mysql::getBinlogStatus(subscription)

// 查询当前用户可见且尚未释放的订阅。
allStatus = mysql::getBinlogStatus()

select subscriptionName, state, rowsWritten, lastError from status

语法

mysql::close(conn)

详情

断开连接，关闭 MySQL 句柄。

参数

conn 通过 mysql::connect 获得的 MySQL 连接句柄。

示例

mysql::close(conn)

• DolphinDB 中数值类型均为有符号类型。为了防止溢出，所有无符号类型会被转化为高一阶的有符号类型。例如，无符号 CHAR 转化为有符号 SHORT，无符号 SHORT 转化为有符号 INT，等等。64 位无符号类型不予支持。
• DolphinDB 不支持 unsigned long long 类型。若 MySQL 中的类型为 bigint unsigned, 可在 load 或者 loadEx 的 schema 参数中设置为 DOUBLE 或者 FLOAT。
• DolphinDB 中各类整形的最小值为 NULL 值，如 CHAR 的-128，SHORT 的-32,768，INT 的-2,147,483,648以及 LONG 的-9,223,372,036,854,775,808。

注：

• IEEE754 浮点数类型皆为有符号数。
• 浮点类型 float 和 double 可转化为 DolphinDB 中的数值相关类型(BOOL, CHAR, SHORT, INT, LONG, FLOAT, DOUBLE)。
• newdecimal/decimal 类型目前仅可转化为 DOUBLE。

以上类型皆可转化为 DolphinDB 中的时间相关类型(DATE, MONTH, TIME, MINUTE, SECOND, DATETIME, TIMESTAMP, NANOTIME, NANOTIMESTAMP)。

长度不超过 10 的 char 和 varchar 将被转化为 SYMBOL 类型，其余转化为 STRING 类型。

string 类型可以转化为转化为 DolphinDB 中的字符串相关类型(STRING, SYMBOL)。

enum 类型可以转化为 DolphinDB 中的字符串相关类型(STRING, SYMBOL)，默认转化为 SYMBOL 类型。

• CPU：i7-7700 3.60GHZ
• 硬盘：SSD，读速为每秒 460~500MB。

美国股票市场从 1990 年至 2016 年的每日数据，共 50,591,907 行，22 列，6.5GB。 导入耗时 160.5秒。