Confluent Platform: ksqlDB数据转换( join查询）

淡淡的烟草味﹌ 2022-08-30 15:00 8阅读 0赞

### 文章目录 ###

*  1, mysql数据源准备
 *  2, ksqldb创建stream

# 1, mysql数据源准备 #

create table employee (id int, dept_id int, name varchar(20) );
    create table dept ( id int,  name varchar(20));
    create table salary ( id int, emp_id int, money DOUBLE );
    
    INSERT INTO test.employee VALUES(1, 1, 'Li si');
    INSERT INTO test.employee VALUES(2, 2, 'Test 1');
    INSERT INTO test.employee VALUES(3, 2, 'Zheng ze');
    
    INSERT INTO test.dept VALUES(1, 'java');
    INSERT INTO test.dept VALUES(2, 'py');
    INSERT INTO test.dept VALUES(3, 'shell');
    
    INSERT INTO test.salary VALUES(1, 1, 18880.23);
    INSERT INTO test.salary VALUES(2, 2, 22233.0);
    INSERT INTO test.salary VALUES(3, 3, 333221.98);
    
    //关联查询
    select emp.* , dept.name as deptname , salary.money from employee emp
    inner join dept on dept.id=emp.dept_id
    inner join salary on salary.emp_id= emp.id
    
    +------+---------+----------+----------+-----------+
    | id   | dept_id | name     | deptname | money     |
    +------+---------+----------+----------+-----------+
    |    1 |       1 | Li si    | java     |  18880.23 |
    |    2 |       1 | Test 1   | java     |     22233 |
    |    3 |       2 | Zheng ze | py       | 333221.98 |
    +------+---------+----------+----------+-----------+

# 2, ksqldb创建stream #

*  测试发现： confluent5.x 只支持单个join （如出现join两次会报错）, confluent 6.x 是支持多次join的  
    ![在这里插入图片描述][c021f823cc51ef08b91531d733622f60.png]

ksq> CREATE SOURCE CONNECTOR `mysql3-dept` WITH(
        "connector.class"='io.confluent.connect.jdbc.JdbcSourceConnector',
        "connection.url"='jdbc:mysql://192.168.56.1:3306/test?user=root&password=root',
        "mode"='incrementing',
        "incrementing.column.name"='id',
        "topic.prefix"='mysql3-',
        "table.whitelist"='dept',
        "key"='id');
    
    CREATE SOURCE CONNECTOR `mysql3-employee` WITH(
        "connector.class"='io.confluent.connect.jdbc.JdbcSourceConnector',
        "connection.url"='jdbc:mysql://192.168.56.1:3306/test?user=root&password=root',
        "mode"='incrementing',
        "incrementing.column.name"='id',
        "topic.prefix"='mysql3-employee',
        "validate.non.null"=false,  
        "query"='select id , dept_id, name from employee',
        "key"='id');
    
    CREATE SOURCE CONNECTOR `mysql3-salary` WITH(
        "connector.class"='io.confluent.connect.jdbc.JdbcSourceConnector',
        "connection.url"='jdbc:mysql://192.168.56.1:3306/test?user=root&password=root',
        "mode"='incrementing',
        "incrementing.column.name"='id',
        "topic.prefix"='mysql3-',
        "table.whitelist"='salary',
        "validate.non.null"=false,
        "key"='id');
    
    create stream dept with( kafka_topic='mysql3-dept',value_format='avro') ;
    create stream employee with( kafka_topic='mysql3-employee',value_format='avro') ;
    create stream salary with( kafka_topic='mysql3-salary',value_format='avro') ;
    
    CREATE STREAM emp_dept_salary3 
    WITH (
        KAFKA_TOPIC = 'emp_dept_salary_topic3',
        VALUE_FORMAT = 'AVRO',
        PARTITIONS = 1,
        REPLICAS = 1   
    ) AS 
    select  emp.id , emp.name,  salary.money from employee emp
    inner join salary WITHIN 60 minutes on salary.emp_id= emp.id
    emit changes;
    
    #mysql插入数据后, stream数据会实时变化
    ksql> select * from emp_dept_salary2 emit changes;
    +--------------+--------------+--------------+--------------+--------------+
    |ROWTIME       |ROWKEY        |EMP_ID        |NAME          |MONEY         |
    +--------------+--------------+--------------+--------------+--------------+
    |1626835283414 |1             |1             |Li si         |18880.23      |
    |1626835283416 |2             |2             |Test 1        |22233.0       |
    |1626835283416 |3             |3             |Zheng ze      |333221.98     |
    |1626835283416 |4             |4             |Ssdf s        |123121.0      |
    |1626835283417 |5             |5             |a sdfxxxxxx   |34522.0       |
    |1626835283417 |6             |6             |XX            |623112.0      |
    |1626837164478 |11            |11            |emp11         |5234233.0     |
    |1626837399610 |12            |12            |e12           |44123.0       |
    
    /* confluent 6.2 ksql> CREATE STREAM join3 WITH ( KAFKA_TOPIC = 'join3', VALUE_FORMAT = 'AVRO', PARTITIONS = 1, REPLICAS = 1 ) AS select emp.id, emp.name , emp.dept_id, salary.money,dept.name from employee emp left join salary WITHIN 60 minutes on salary.emp_id=emp.id left join dept WITHIN 60 minutes on emp.dept_id=dept.id emit changes; ksql> select * from JOIN3 EMIT CHANGES; +--------------+--------------+--------------+--------------+--------------+ |DEPT_ID |EMP_ID |EMP_NAME |MONEY |DEPT_NAME | +--------------+--------------+--------------+--------------+--------------+ |1 |1 |Li si |18880.23 |java | |1 |2 |Test 1 |22233.0 |java | |2 |3 |Zheng ze |333221.98 |py | |3 |4 |Ssdf s |123121.0 |shell | |2 |5 |a sdfxxxxxx |34522.0 |py | |3 |6 |XX |623112.0 |shell | |8 |8 |tt |null |null | */

[c021f823cc51ef08b91531d733622f60.png]: /images/20220829/35070db7ddc74ad0bfddff8d2e463ba0.png