-- 创建⼀个名为ecommerce_db的数据库，⽤于存储电商⽹站的数据CREATE DATABASE IF NOT EXISTS ecommerce_db;USE ecommerce_db;-- 创建⼀个名为orders的表，⽤于存储订单信息CREATE TABLE IF NOT EXISTS orders ( id INT AUTO_INCREMENT PRIMARY KEY, -- ⾃增主键 order_id VARCHAR(50) NOT NULL, -- 订单编号，不能为空 user_id INT NOT NULL, -- ⽤户ID，不能为空 product_id INT NOT NULL, -- 产品ID，不能为空 quantity INT NOT NULL, -- 订购数量，不能为空 order_date TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP -- 订单⽇期，默认值为当前时间戳，不能为空);-- 插⼊⼀些测试数据到orders表INSERT INTO orders (order_id, user_id, product_id, quantity) VALUES ('ORD123', 1, 101, 2), ('ORD124', 2, 102, 1), ('ORD125', 3, 103, 3), ('ORD126', 1, 104, 1), ('ORD127', 2, 105, 5);

  //  

CREATE TABLE binlog_source ( id int, order_id STRING, user_id INT, product_id int, quantity int, order_date TIMESTAMP(3) ) WITH ( 'connector' = 'binlog-x', 'username' = 'root', 'password' = 'root', 'cat' = 'insert,delete,update', 'url' = 'jdbc:mysql://localhost:3306/ecommerce_db?useSSL=false', 'host' = 'localhost', 'port' = '3306', 'table' = 'ecommerce_db.orders', 'timestamp-format.standard' = 'SQL', 'scan.parallelism' = '1' ); CREATE TABLE kafka_sink ( id int, order_id STRING, user_id INT, product_id int, quantity int, order_date TIMESTAMP(3),PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'upsert-kafka-x', 'topic' = 'orders', 'properties.bootstrap.servers' = 'localhost:9092', 'key.format' = 'json', 'value.format' = 'json', 'value.fields-include' = 'ALL', 'sink.parallelism' = '1' ); insert into kafka_sink select * from binlog_source u;

CREATE TABLE kafka_source ( id int, order_id STRING, user_id INT, product_id INT, quantity INT, order_date TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'upsert-kafka-x', 'topic' = 'orders', 'properties.bootstrap.servers' = 'localhost:9092', 'properties.group.id' = 'test_group', 'key.format' = 'json', 'value.format' = 'json', 'scan.parallelism' = '1' ); CREATE TABLE hbase_sink( rowkey STRING, order_info ROW < order_id STRING, user_id STRING, product_id STRING, quantity STRING, order_date STRING >, PRIMARY KEY (rowkey) NOT ENFORCED ) WITH( -- 这⾥以hbase14为例，如果hbase版本是2.x，我们可以使⽤hbase2-x插件代替 'connector' = 'hbase14-x', 'zookeeper.quorum' = 'localhost:2181', 'zookeeper.znode.parent' = '/hbase', 'table-name' = 'ecommerce_db:orders', 'sink.parallelism' = '1' ); INSERT INTO hbase_sink SELECT cast(id as STRING), ROW( cast(order_id as STRING), cast(user_id as STRING), cast(product_id as STRING), cast(quantity as STRING), cast(order_date as STRING) ) FROM kafka_source

  //  

  //  

CREATE TABLE order.realtime_order_archive ( order_id INT PRIMARY KEY COMMENT '订单唯⼀标识', customer_id INT COMMENT '客户唯⼀标识', product_id INT COMMENT '产品唯⼀标识', order_date TIMESTAMP COMMENT '订单⽇期和时间', payment_method VARCHAR(255) COMMENT '⽀付⽅式（信⽤卡、⽀付宝、微信⽀付等）', shipping_method VARCHAR(255) COMMENT '配送⽅式（顺丰速运、圆通速递等）', shipping_address VARCHAR(255) COMMENT '配送地址', order_total DECIMAL(10,2) COMMENT '订单总⾦额', discount DECIMAL(10,2) COMMENT '折扣⾦额', order_status VARCHAR(255) COMMENT '订单状态（已完成、已取消等）' ); 

"name": "mysqlreader", "parameter": { "column" : [ "*" //这⾥假设我们读取所有字段，可以填写‘*’ ], "increColumn": "id", "polling": true, "pollingInterval": 3000, "username": "username", "password": "password", "connection": [ { "jdbcUrl": [ "jdbc:mysql://ip:3306/liuliu?useSSL=false" ], "schema":"order", "table": [ "realtime_order_archive" ] } ] } }