MatrixOne 支持多样化生态工具，持续提升开发者体验

作者：MatrixOrigin

2023-11-16
上海
本文字数：3537 字
阅读完需：约 12 分钟

概述

在选择一款数据库产品时，对数据库上下游生态组件的打通是大数据开发工程师需要面对的一致难题。

MatrixOne 提出了“One Size Fits Most”理念，旨在用全新 HSTAP 技术架构打破数据孤岛，其中，与各生态组件的“无缝衔接”也是 MatrixOne“一栈式”愿景中的重要一环。

MatrixOne 高度兼容 MySQL 协议，在 1.0 正式版发布前，MatrixOne 收集整理了社区反馈较多的上下游组件并进行了更为完整的适配。本轮适配主要围绕：数据集成工具、BI 工具和数据计算引擎这三类生态工具。

Part 1 数据集成工具

数据集成工具可以将来自不同数据源的数据整合为统一的数据集，通常具备从不同数据源中收集数据、对数据进行清洗、转换、重构和整合的能力，以便能够在一个统一的数据仓库或数据湖中进行存储和管理。

与数据集成工具的适配可帮助数据开发工程师更丝滑高效的进行初期存量数据的迁移和后续日常的 ETL 作业，极大的保障工作效率和业务的稳定性。

MatrixOne 在征集社区建议后，本轮优先对 DataX 与 SeaTunnel 进行了适配。得益于对 MySQL 协议的高度兼容，经过本轮适配，我们在 DataX 或 SeaTunnel 中可以直接使用 MySQL 所对应的 Writer 或 Driver 来对 MatrixOne 进行连接访问。

在 DataX 官方版本中，mysqlwriter 默认使用版本较老的 mysql-connector-java-5.1.47.jar 驱动，为保证拥有更好的性能和兼容性，在适配了 mysqlwriter 的基础上，MatrixOne 社区还基于 mysqlwriter 改造出使用 mysql-connector-j-8.0.33.jar 驱动的 matrixonewriter，作业 json 示例如下，欢迎社区同学下载使用：

{    "job": {        "setting": {            "speed": {                "channel": 1            }        },        "content": [            {                 "reader": {                    "name": "streamreader",                    "parameter": {                        "column" : [                            {                                "value": 20210106,                                "type": "long"                            },                            {                                "value": "matrixone",                                "type": "string"                            }                        ],                        "sliceRecordCount": 1000                    }                },                "writer": {                    "name": "matrixonewriter",                    "parameter": {                        "writeMode": "insert",                        "username": "root",                        "password": "111",                        "column": [                            "M_ID",                            "M_NAME"                        ],                        "preSql": [                            "delete from m_user"                        ],                        "connection": [                            {                                "jdbcUrl": "jdbc:mysql://127.0.0.1:6001/mo_demo",                                "table": [                                    "m_user"                                ]                            }                        ]                    }                }            }        ]    }}

复制代码

Part 2 BI 工具

BI 即指商业智能（Business Intelligence），指用现代数据仓库技术、线上分析处理技术、数据挖掘和数据展现技术进行数据分析以实现商业价值。例如数据分析师会使用 BI 工具，基于海量的散乱数据进行数据分析，最终产出清晰的、可用于支持决策的数据报表。

BI 工具所用的数据通常存储在数据库中，由于不同数据库之间语法存在不同程度的差异，BI 厂商通常会与数据库厂商进行兼容性互认并针对性的开发出对应的连接器。

MatrixOne 本次针对 Superset、FineBI 和永洪 BI 进行了首轮的适配，当我们在使用上述 BI 工具连接 MatrixOne 时，可以直接选择 MySQL 连接器，将 MatrixOne 视为 MySQL 进行连接访问，来对用户的数据可视化场景提供有效支撑。下图为永洪 BI 使用 MySQL 数据源连接示例：

BI 的适配工作无法做到“一步到位”，MatrixOne 当前先基于 MySQL 语法兼容性来实现与 BI 的高度兼容，随着 MatrixOne 新版本个性化能力的不断扩展，我们也会积极推进和 BI 厂商的互认，来协同开发出兼容性更优的专用连接器。

Part 3 数据计算引擎

大数据计算引擎主要用于对数据的加工、清洗、计算或迁移导入，常用的计算引擎例如：MapReduce、Storm、Spark、Flink 等。其中，Spark 和 Flink 作为目前主流的批量/流式计算引擎，MatrixOne 进行了细节的测试和适配。

MatrixOne 规划中的“流引擎”能力在持续完善中，在适配 Spark/Flink 引擎后，开发者可以直接使用通用 JDBC 驱动，在 Spark 或 Flink 中通过外部算力进行大数据量级的数据清洗，再或着，通过 Flink 实现更灵活的“流式数据源”的数据接入。例如，MatrixOne 1.0 版本原生实现了与 Kafka 中数据源的连接：

CREATE SOURCE | SINK CONNECTOR [IF NOT EXISTS] connector_name CONNECTOR_TYPE WITH（property_name = expression [, ...]）;

复制代码

通过本轮适配 Flink 引擎，MatrixOne 即可扩展支持包括 Kafka、Pulsar、RocketMQ 等更广泛的消息队列产品。以使用 Flink Table API 为例：

--定义Kafka源表CREATE TABLE kafka_source (  `user_id` BIGINT,  `item_id` BIGINT,  `behavior` STRING) WITH (  'connector' = 'kafka',  'topic' = 'user_behavior',  'properties.bootstrap.servers' = 'localhost:9092',  'properties.group.id' = 'testGroup',  'scan.startup.mode' = 'earliest-offset',  'format' = 'csv')
--定义MatrixOne目标表，驱动使用mysql驱动CREATE TABLE matrixone_kafka_sink (  `user_id` BIGINT,  `item_id` BIGINT,  `behavior` STRING,   PRIMARY KEY(user_id) NOT ENFORCED) WITH (  'connector' = 'mysql',  'hostname' = '<yourHostname>',  'port' = '6001',  'username' = '<yourUsername>',  'password' = '<yourPassword>',  'database-name' = '<yourDatabaseName>',  'table-name' = '<yourTableName>');
--同步Kafka流式数据到MatrixOneINSERT INTO matrixone_sink SELECT * FROM kafka_source;

复制代码

此外，借助 Flink CDC，MatrixOne 也可支持对传统 TP 类数据库的数据同步，这项能力的补齐可帮助企业在不改变原业务上游框架的基础上扩展加入 MatrixOne，然后使用 MatrixOne 统一下游业务所需的数据分析场景，真正实现“Single Source of Truth”。仍以 Flink Table API 为例：

--定义MySQL CDC流式数据源CREATE TABLE mysql_cdc_orders (  `user_id` BIGINT,  `item_id` BIGINT,  `behavior` STRING,  PRIMARY KEY(user_id) NOT ENFORCED) WITH (  'connector' = 'mysql-cdc',  'hostname' = 'localhost',  'port' = '3306',  'username' = 'flinkuser',  'password' = 'flinkpw',  'database-name' = 'mydb',  'table-name' = 'orders');
--定义MatrixOne目标表，驱动使用mysql驱动CREATE TABLE matrixone_cdc_sink (  `user_id` BIGINT,  `item_id` BIGINT,  `behavior` STRING,   PRIMARY KEY(user_id) NOT ENFORCED) WITH (  'connector' = 'mysql',  'hostname' = '<yourHostname>',  'port' = '6001',  'username' = '<yourUsername>',  'password' = '<yourPassword>',  'database-name' = '<yourDatabaseName>',  'table-name' = '<yourTableName>');
--同步MySQL binlog流式数据到MatrixOneINSERT INTO matrixone_cdc_sink SELECT * FROM mysql_cdc_orders;

复制代码

随着 MatrixOne 1.0 正式版发版脚步渐近，越来越多的小伙伴开始在更多的业务中测试和使用 MatrixOne，也为社区提出了非常多的宝贵建议。MatrixOne 社区将响应社区开发者的呼声，持续适配主流的生态组件，不断提升产品的易用性和友好性。欢迎大家反馈在使用过程中发现的兼容性问题，您的每一个 issue 我们都无比珍视，社区开发者将会第一时间进行排期修复。