Flinksql 无法写入mysql

要将实时计算的当天 PV 写入 MySQL，可以使用 Flink 的 JDBC Connector 连接器，将结果输出到 MySQL 数据库中。 以下是一个示例代码： ``` DataStream<Tuple2<String, Integer>> pvStream = ...; // 获取实时计算的当天 PV 数据流 // 将结果写入 MySQL pvStream.addSink(JdbcSink.sink( "INSERT INTO pv_count(date, pv) VALUES (?, ?)", (ps, t) -> { ps.setDate(1, new java.sql.Date(System.currentTimeMillis())); ps.setInt(2, t.f1); }, new JdbcConnectionOptions.JdbcConnectionOptionsBuilder() .withUrl("jdbc:mysql://localhost:3306/db") .withDriverName("com.mysql.jdbc.Driver") .withUsername("user") .withPassword("password") .build() )); ``` 其中，`pvStream` 是一个包含了当天 PV 数量的数据流。`JdbcSink.sink` 方法将数据写入 MySQL 数据库中，第一个参数是 SQL 语句，第二个参数是将数据写入 PreparedStatement 中的逻辑，第三个参数是连接 MySQL 的配置信息。 在这个示例中，我们使用了一个 `pv_count` 表，用于存储每天的 PV 数量。表中包含两个字段：`date` 用于存储日期，`pv` 用于存储当天的 PV 数量。在 `JdbcSink.sink` 的第一个参数中，我们使用了一个占位符 `?` 来表示这两个字段。在第二个参数中，我们将当前时间作为日期写入了 PreparedStatement 中的第一个参数，将当天 PV 数量写入了第二个参数。在第三个参数中，我们指定了连接 MySQL 数据库的配置信息。 这样，当 Flink 实时计算出当天的 PV 数量后，就会自动将结果写入 MySQL 数据库中。

### 使用 Flink SQL 实现从 MySQL 到另一个 MySQL 的数据处理或迁移 为了实现从一个 MySQL 数据库到另一个 MySQL 数据库的数据处理或迁移，可以利用 Apache Flink 提供的流式 SQL 功能以及其内置的连接器（Connector）。以下是具体方法： #### 1. 配置源表（Source Table） Flink 支持通过 `CREATE TABLE` 声明一个基于 MySQL Binlog 的源表。此表会捕获 MySQL 中的变化并将其作为变更日志流进行消费。 ```sql CREATE TABLE source_table ( id BIGINT, name STRING, age INT, PRIMARY KEY (id) NOT ENFORCED -- 定义主键以便优化操作 ) WITH ( 'connector' = 'mysql-cdc', -- 使用 Debezium CDC 来监控 MySQL 变更 'hostname' = 'source-mysql-host', 'port' = '3306', 'username' = 'root', 'password' = 'your-password', 'database-name' = 'db_name', 'table-name' = 'table_name' ); ``` 上述配置定义了一个名为 `source_table` 的表，它能够实时捕捉来自指定 MySQL 表中的增删改操作[^1]。 #### 2. 处理逻辑（Transformation Logic） 在实际应用中可能需要对原始数据执行一些转换或者过滤操作。这些可以通过标准 SQL 查询完成。 例如，假设我们只希望同步年龄大于等于18岁的记录，则可编写如下查询语句： ```sql SELECT * FROM source_table WHERE age >= 18; ``` 如果涉及复杂计算比如聚合函数求平均数等也可以在此阶段加入相应表达式[^2]。 #### 3. 配置目标表（Sink Table） 同样地，在接收端也需要创建一张对应的 Sink 表用于存储最终结果集。 ```sql CREATE TABLE sink_table ( id BIGINT, name STRING, age INT ) WITH ( 'connector' = 'jdbc', -- JDBC connector for writing into another MySQL instance. 'url' = 'jdbc:mysql://target-mysql-host:3306/db_name?useSSL=false&serverTimezone=UTC', 'table-name' = 'destination_table', 'driver' = 'com.mysql.cj.jdbc.Driver', 'username' = 'root', 'password' = 'another-password' ); ``` 这里指定了一个新的 MySQL 实例作为目标位置，并设置了必要的参数以建立连接[^3]。 #### 4. 启动作业并将数据写入目标表 最后一步就是启动整个 ETL 流程，即将经过筛选后的数据插入至目标表当中去。 ```sql INSERT INTO sink_table SELECT * FROM source_table WHERE age >= 18; ``` 这条命令将会持续不断地把符合条件的新条目追加到远程服务器上的对应关系型结构里直到手动停止为止^。 ### 注意事项 - **性能调优**：对于大规模生产环境而言，应该考虑调整批大小、并发度等相关设置来提升吞吐量。 - **错误恢复机制**：确保启用了 checkpointing 和 savepoints ，这样即使发生意外中断也能快速恢复正常运行状态而无需重新开始全部过程。 ```python # Example Python code snippet showing how you might configure checkpoints programmatically within a PyFlink application. from pyflink.datastream import StreamExecutionEnvironment from pyflink.table import StreamTableEnvironment env = StreamExecutionEnvironment.get_execution_environment() t_env = StreamTableEnvironment.create(env) # Enable checkpoint every 5 minutes with interval of 30 seconds between attempts if failure occurs during saving progress information. env.enable_checkpointing(300000, mode="EXACTLY_ONCE") env.get_checkpoint_config().set_min_pause_between_checkpoints(30000) ```