Flink的SqlClient工具

左手的ㄟ右手 2022-11-07 14:53 120阅读 0赞

## SqlClient ##

启动换一个yarn-session,

bin/yarn-session -d

然后启动一个sql客户端.

bin/sql-client.sh embedded

### 建立到Kafka的连接 ###

下面创建一个流表从Kafka读取数据

copy 依赖到 flink的lib 目录下 flink-sql-connector-kafka\_2.11-1.12.0.jar 下载地址: [https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka\_2.11/1.12.0/flink-sql-connector-kafka\_2.11-1.12.0.jar][https_repo.maven.apache.org_maven2_org_apache_flink_flink-sql-connector-kafka_2.11_1.12.0_flink-sql-connector-kafka_2.11-1.12.0.jar]

create table sensor(id string, ts bigint, vc int)
    with(
        'connector'='kafka',
        'topic'='flink_sensor',
        'properties.bootstrap.servers'='hadoop162:9092',
        'properties.group.id'='atguigu',
        'format'='json',
        'scan.startup.mode'='latest-offset'
    )

从流表查询数据

select * from sensor;

![在这里插入图片描述][20210312230810573.png]

向Kafka写入数据: `{"id": "sensor1", "ts": 1000, "vc": 10}`  
![在这里插入图片描述][20210312230824369.png]

### 建立到mysql的连接 ###

依赖: flink-sql-connector-kafka_2.11-1.12.0.jar

copy mysql驱动到lib目录

create table sensor(id string, ts bigint, vc int)
    with(
        'connector' = 'jdbc',
        'url' = 'jdbc:mysql://hadoop162:3306/gmall',
        'username'='root',
        'password'='aaaaaa',
        'table-name' = 'sensor'
    )

[https_repo.maven.apache.org_maven2_org_apache_flink_flink-sql-connector-kafka_2.11_1.12.0_flink-sql-connector-kafka_2.11-1.12.0.jar]: https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka_2.11/1.12.0/flink-sql-connector-kafka_2.11-1.12.0.jar
[20210312230810573.png]: /images/20221023/7d7cb2677e9645b9a788b1c1de3a4ba6.png
[20210312230824369.png]: /images/20221023/cd7f7ab585f847159a804a91779d7323.png