使用flink sql 读取kafka topic 写入mysql 1.13
时间: 2024-01-20 21:09:33 浏览: 96
好的,您可以按照以下步骤操作:
1. 在 Flink 中添加 Kafka 和 MySQL 的依赖:
```xml
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>8.0.23</version>
</dependency>
```
2. 创建 Flink SQL 的执行环境:
```java
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
EnvironmentSettings settings = EnvironmentSettings.newInstance()
.useBlinkPlanner()
.inStreamingMode()
.build();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
```
3. 注册 Kafka 数据源和 MySQL 数据汇:
```java
tableEnv.executeSql("CREATE TABLE kafka_source (\n" +
" id INT,\n" +
" name STRING,\n" +
" age INT,\n" +
" PRIMARY KEY (id) NOT ENFORCED\n" +
") WITH (\n" +
" 'connector' = 'kafka',\n" +
" 'topic' = 'test',\n" +
" 'properties.bootstrap.servers' = 'localhost:9092',\n" +
" 'properties.group.id' = 'testGroup',\n" +
" 'format' = 'json',\n" +
" 'scan.startup.mode' = 'earliest-offset'\n" +
")");
tableEnv.executeSql("CREATE TABLE mysql_sink (\n" +
" id INT,\n" +
" name STRING,\n" +
" age INT,\n" +
" PRIMARY KEY (id)\n" +
") WITH (\n" +
" 'connector' = 'jdbc',\n" +
" 'url' = 'jdbc:mysql://localhost:3306/test',\n" +
" 'table-name' = 'user',\n" +
" 'driver' = 'com.mysql.cj.jdbc.Driver',\n" +
" 'username' = 'root',\n" +
" 'password' = 'root'\n" +
")");
```
4. 使用 Flink SQL 读取 Kafka 数据源并将数据写入 MySQL 数据汇:
```java
tableEnv.executeSql("INSERT INTO mysql_sink SELECT * FROM kafka_source");
env.execute();
```
这样就可以使用 Flink SQL 从 Kafka 中读取数据,并将数据写入 MySQL 数据库中了。