Flink自定義Sink

Flink自定義Sink

Flink 自定義Sink，把socket數據流數據轉換成對象寫入到mysql存儲。

#創建Student類
public class Student {

    private int id;
    private String name;
    private int age;

    @Override
    public String toString() {
        return "Student{" +
                "id=" + id +
                ", name='" + name + '\'' +
                ", age=" + age +
                '}';
    }

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }
}

Step2:繼承RichSinkFunction，自定義Sink

public class Sink2Mysql extends RichSinkFunction<Student> {

    Connection connection;
    PreparedStatement pstmt;

    private Connection getConnection() {
        Connection conn = null;
        try {
            Class.forName("com.mysql.jdbc.Driver");
            String url = "jdbc:mysql://localhost:3306/imooc_flink";
            conn = DriverManager.getConnection(url,"root","123456");

        } catch (Exception e) {
            e.printStackTrace();
        }

        return conn;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        connection = getConnection();
        String sql = "insert into student(id,name,age) values (?,?,?)";
        pstmt = connection.prepareStatement(sql);


        System.out.println("open");
    }

    // 每條記錄插入時調用一次
    public void invoke(Student value, Context context) throws Exception {
        System.out.println("invoke~~~~~~~~~");
        // 未前面的占位符賦值
        pstmt.setInt(1, value.getId());
        pstmt.setString(2, value.getName());
        pstmt.setInt(3, value.getAge());

        pstmt.executeUpdate();

    }

    @Override
    public void close() throws Exception {
        super.close();

        if(pstmt != null) {
            pstmt.close();
        }

        if(connection != null) {
            connection.close();
        }
    }
}

Step3：在mysql創建存儲表

create table student (
id int(11) not null auto_increment,
name varchar(25),
age int(10),
primary key (id)
);

Step4：
將socket流轉成Student對象，並存儲到mysql

public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    DataStreamSource<String> source =  env.socketTextStream("localhost",7777);

    SingleOutputStreamOperator<Student> stuStream = source.map(new MapFunction<String, Student>() {
        @Override
        public Student map(String value) {
            String[] splits =  value.split(",");

            Student stu = new Student();
            stu.setId(Integer.parseInt(splits[0])) ;
            stu.setName(splits[1]);
            stu.setAge(Integer.parseInt(splits[2]));

            return stu;
        }
    });

    stuStream.addSink(new Sink2Mysql());

    env.execute("JavaStreamSink2MysqlApp");
}

Step5：測試
在終端開啟socket流，並輸入數據：
1,xiao,17
2,ming,24
3,uzi,20

查詢mysql表：
select * from student;

結果如下：

mysql> select * from student;
+----+------+------+
| id | name | age  |
+----+------+------+
|  1 | xiao |   17 |
|  2 | ming |   24 |
|  3 | uzi  |   20 |
+----+------+------+
3 rows in set (0.00 sec)

注意一點：
如果運行程序時，報錯：
java.lang.ClassNotFoundException: com.mysql.jdbc.Driver

可能原因是沒有添加依賴 mysql-jdbc 依賴，需要在pom文件添加：

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.47</version>
</dependency>

這里的版本不需要跟mysql版本對應。