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

{    "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"                                ]                            }                        ]                    }                }            }        ]    }}

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

--定义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;

--定义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;