/*******************************************************************************
 * Package: com.song.bigdata.stream
 * Type:    JdbcMysqlWrite
 * Date:    2022-10-30 14:49
 *
 *
 *
 * You may not use this file except in compliance with the License.
 *******************************************************************************/
package com.song.bigdata.stream;

import com.song.bigdata.pojo.User;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.concurrent.TimeUnit;

/**
 * 功能描述：
 *
 * @author Songxianyang
 * @date 2022-10-30 14:49
 */
public class JdbcMysqlWrite {
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                3, // 尝试重启的次数
                Time.of(10, TimeUnit.SECONDS) // 间隔
        ));
        // 并行度
        environment.setParallelism(1);
        DataStreamSource<User> streamSource = environment.fromElements(
                new User("song-user", 12),
                new User("xian-user", 92),
                new User("song-user", 11),
                new User("song-user", 110),
                new User("song-user", 90),
                new User("xian-user", 10)
        );
        DataStreamSink<String> root = streamSource.map(new MapFunction<User, String>() {
            @Override
            public String map(User user) throws Exception {
                return user.name;
            }
        }).addSink(JdbcSink.sink(
                // ? 问好千万别写成 中文的 ？ 否者报异常：Parameter index out of range (1 > number of parameters, which is 0).
                "INSERT INTO b (value) VALUES (?)", new JdbcStatementBuilder<String>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, String s) throws SQLException {
                        preparedStatement.setString(1, s);
                    }
                }, new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withUrl("jdbc:mysql://localhost:3306/person?characterEncoding=UTF-8&useUnicode=true&useSSL=false&tinyInt1isBit=false&allowPublicKeyRetrieval=true&serverTimezone=Asia/Shanghai")
                        .withDriverName("com.mysql.cj.jdbc.Driver").withUsername("root").withPassword("root#123").build()
        ));
        environment.execute();
    }

}