java Flink使用addSink方法保存流到mysql数据库中_flink的addsink(mysql)

博主把核心的内容写在最前面，其他内容和完整的代码放在最后面哈：

pom配置

      <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.11</version>
        </dependency>
1
2
3
4
5

主要代码

package write_to_mysql;

import create_data.MyData2; // 格式见其他内容
import create_data.MyDataSource2; // 格式见其他内容
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


public class GetMain {
    public static void main(String[] args) throws Exception {
        // set up the streaming execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<MyData2> sourceStream = env.addSource(new MyDataSource2()); // 得到数据源
        sourceStream.addSink(new MysqlSink()); // 核心！保存到mysql
        env.execute("Flink_to_mysql demo");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

可以看到，使用流.addSink()就可以保存流的数据了，这个MysqlSink是自己写的保存逻辑，代码如下：

package write_to_mysql;

import create_data.MyData2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class MysqlSink extends RichSinkFunction<MyData2> {
    private PreparedStatement ps = null;
    private Connection connection = null;
    String driver = "com.mysql.jdbc.Driver";
    String url = "jdbc:mysql://127.0.0.1:3306/my_test_db?useSSL=false";
    String username = "testuser";
    String password = "testpassword";

    @Override
    public void open(Configuration parameters) throws Exception { // 要执行的代码
        super.open(parameters);  // 用于建立连接
        Class.forName(driver);  //加载JDBC驱动
        connection = DriverManager.getConnection(url, username, password);
        String sql = "insert into test_db.test_csv (col_1,col_2,col_3,col_4)" +
                "values (?,?,?,'what');";
        ps = connection.prepareStatement(sql);
    }

    @Override
    public void invoke(MyData2 value, Context context) throws Exception { // 真正执行的操作
        // 每次插入都会调用一次
        // 这里的value.xxx根据具体的操作逻辑来
        // ps.setxxx(n,xxx) 这里的n代表要保存的位置，也就是要把数据拍到上面的String sql的第几个?(问号)上
        ps.setString(1, String.valueOf(value.keyId));
        ps.setString(2, String.valueOf(value.timestamp));
        ps.setString(3, String.valueOf(value.num));
        ps.executeUpdate();
    }

    @Override
    public void close() throws Exception { // 关闭操作
        super.close();
        if (connection != null) {
            connection.close();
        }
        if (ps != null) {
            ps.close();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

只需要覆写open，invoke，close三个函数即可，一个用于打开连接，一个用于执行操作，一个用于关闭连接。

其他内容：MyData2类，与生成数据源的类MyDataSource2

数据类与生成数据的类请参考：https://blog.csdn.net/weixin_35757704/article/details/120626180

MyData2.java

package create_data;

import java.util.Arrays;

public class MyData2 {
    public int keyId;
    public long timestamp;
    public int num;
    public double[] valueList;

    public MyData2() {
    }

    public MyData2(int accountId, long timestamp, int num, double[] valueList) {
        this.keyId = accountId;
        this.timestamp = timestamp;
        this.num = num;
        this.valueList = valueList;
    }

    public long getKeyId() {
        return keyId;
    }

    public void setKeyId(int keyId) {
        this.keyId = keyId;
    }

    public long getTimestamp() {
        return timestamp;
    }

    public void setTimestamp(long timestamp) {
        this.timestamp = timestamp;
    }

    public double[] getValueList() {
        return valueList;
    }

    public void setValueList(double[] valueList) {
        this.valueList = valueList;
    }

    public int getNum() {
        return num;
    }

    public void setNum(int num) {
        this.num = num;
    }

    @Override
    public String toString() {
        return "MyData{" +
                "keyId=" + keyId +
                ", timestamp=" + timestamp +
                ", num=" + num +
                ", valueList= " + Arrays.toString(valueList) +
                '}';
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

MyDataSource2.java

package create_data;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

public class MyDataSource2 implements SourceFunction<MyData2> {
    // 定义标志位，用来控制数据的产生
    private boolean isRunning = true;
    private final Random random = new Random(0);

    @Override
    public void run(SourceContext ctx) throws Exception {
        while (isRunning) {
            ctx.collect(new MyData2(random.nextInt(3), System.currentTimeMillis(), 1, new double[]{random.nextDouble()}));
            Thread.sleep(1000L); // 1s生成1个数据
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24