Flink之JDBCSink连接MySQL

2023-12-15 12:43:20

输出到MySQL

  1. 添加依赖
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-jdbc</artifactId>
  <version>3.1.0-1.17</version>
</dependency>
<dependency>
    <groupId>com.mysql</groupId>
    <artifactId>mysql-connector-j</artifactId>
    <version>8.0.32</version>
</dependency>
  1. 启动MySQL, 在test库下建表clicks
CREATE TABLE `clicks` (
  `user` VARCHAR(100) NOT NULL,
  `url` VARCHAR(100) DEFAULT NULL,
  `ts` BIGINT DEFAULT NULL
) ENGINE=INNODB DEFAULT CHARSET=utf8
  1. 示例代码
public class Flink04_JdbcSink {
    public static void main(String[] args) {
        //1.创建运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //默认是最大并行度
        env.setParallelism(1);

        DataStreamSource<Event> ds = Flink06_EventSource.getEventSource(env);

        //
        SinkFunction<Event> sink = JdbcSink.sink(
                "insert into clicks(user, url, ts) values (?,?,?)"
                , new JdbcStatementBuilder<Event>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, Event event) throws SQLException {
                        //给SQL的占位符赋值
                        preparedStatement.setString(1, event.getUser());
                        preparedStatement.setString(2, event.getUrl());
                        preparedStatement.setLong(3, event.getTs());
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(5)
                        .withBatchIntervalMs(10000)
                        .withMaxRetries(3)
                        .build()
                ,
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUsername("root")
                        .withPassword("000000")
                        .withUrl("jdbc:mysql://hadoop102:3306/flink")
                        .build()
        );

        ds.addSink(sink);

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

MySQL的幂等性处理

  1. 将插入关键字替换为replace,如果主键重复,将除了主键外的所有字段都替换。
  2. 使用on duplicate key update 字段名 = values(字段名)语法,如果主键重复,可以选择部分字段进行替换,其余字段保持不变。
  3. 示例代码
public class Flink05_JdbcSinkReplace {
    public static void main(String[] args) {
        //1.创建运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //默认是最大并行度
        env.setParallelism(1);

        DataStreamSource<Event> ds = Flink06_EventSource.getEventSource(env);

        SingleOutputStreamOperator<WordCount> countDs =
                ds.map(event -> new WordCount(event.getUrl(), 1))
                .keyBy(WordCount::getWord)
                .sum("count");

        //
        SinkFunction<WordCount> sink = JdbcSink.sink(
//                "replace into url_count(url, cnt) values (?,?)"
                "insert into url_count(url, cnt) values(?,?) on duplicate key update cnt = values(cnt)"
                ,
                new JdbcStatementBuilder<WordCount>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, WordCount wordCount) throws SQLException {
                        //注意:这里的起始下标是1
                        preparedStatement.setString(1, wordCount.getWord());
                        preparedStatement.setInt(2, wordCount.getCount());
                    }
                }
                ,
                JdbcExecutionOptions.builder()
                        .withBatchSize(5)
                        .withBatchIntervalMs(10000)
                        .withMaxRetries(3)
                        .build()
                ,
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUsername("root")
                        .withPassword("000000")
                        .withUrl("jdbc:mysql://hadoop102:3306/flink")
                        .build()
        );

        countDs.addSink(sink);

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

文章来源:https://blog.csdn.net/qq_44273739/article/details/134885972
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。