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版本对应。

巴特西

Flink自定义Sink

Flink自定义Sink

最新文章

热门文章