Apache Flink 零基础入门（十七）Flink 自定义Sink

需求：socket发送过来的数据，把String类型转成对象，然后把Java对象保存到Mysql数据库中。

创建数据库和表

create database imooc_flink;
create table student(
id int(11) NOT NULL AUTO_INCREMENT,
name varchar(25),
age int(10),
primary key(id)
)

导入mysql依赖：

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

创建POJO Student

package com.vincent.course05;public class Student {private int id;private String name;private int age;@Overridepublic 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;}
}

然后创建连接，SinkToMySQL.java

package com.vincent.course05;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 SinkToMySQL extends RichSinkFunction<Student> {PreparedStatement ps;private Connection connection;/*** open() 方法中建立连接，这样不用每次 invoke 的时候都要建立连接和释放连接** @param parameters* @throws Exception*/@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);connection = getConnection();String sql = "insert into student(id, name, age) values(?, ?, ?);";ps = this.connection.prepareStatement(sql);}@Overridepublic void close() throws Exception {super.close();//关闭连接和释放资源if (connection != null) {connection.close();}if (ps != null) {ps.close();}}/*** 每条数据的插入都要调用一次 invoke() 方法** @param value* @param context* @throws Exception*/@Overridepublic void invoke(Student value, Context context) throws Exception {//组装数据，执行插入操作ps.setInt(1, value.getId());ps.setString(2, value.getName());ps.setInt(3, value.getAge());ps.executeUpdate();}private static Connection getConnection() {Connection con = null;try {Class.forName("com.mysql.cj.jdbc.Driver");con = DriverManager.getConnection("jdbc:mysql://192.168.152.45:3306/imooc_flink?useUnicode=true&characterEncoding=UTF-8", "root", "123456");} catch (Exception e) {e.printStackTrace();System.out.println("-----------mysql get connection has exception , msg = " + e.getMessage());}return con;}
}

main方法：

public static void main(String[] args) throws Exception {StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> source = environment.socketTextStream("192.168.152.45", 9999);SingleOutputStreamOperator<Student> studentStream = source.map(new MapFunction<String, Student>() {@Overridepublic Student map(String value) throws Exception {String[] splits = value.split(",");Student student = new Student();student.setId(Integer.parseInt(splits[0]));student.setName(splits[1]);student.setAge(Integer.parseInt(splits[2]));return student;}});studentStream.addSink(new SinkToMySQL());environment.execute("JavaCustomSinkToMysql");}

从socket中获取数据，数据格式使用逗号分割，在控制台中输入：

nc -lk 9999
1,tom,23

检查数据库，数据库中多了一条数据

mysql> select * from student;
+----+------+------+
| id | name | age  |
+----+------+------+
|  1 | tom  |   23 |
+----+------+------+
1 row in set (0.00 sec)

这样就很方便的使用自定义的sink，写入到MySQL中去。

总结：

第一步：继承RichSinkFunction<T> T就是想要写入的对象类型

第二步：重写方法 open/close生命周期方法，invoke每条记录执行一次

默认情况下open方法的并行度不是1，跟具体的电脑有关系。

Apache Flink 零基础入门（十七）Flink 自定义Sink相关推荐

Apache Flink 零基础入门【转】
Apache Flink 零基础入门(一):基础概念解析 Apache Flink 零基础入门(二):DataStream API 编程转载于:https://www.cnblogs.com/dav ...
Apache Flink 零基础入门（十八）Flink Table APISQL
什么是Flink关系型API? 虽然Flink已经支持了DataSet和DataStream API,但是有没有一种更好的方式去编程,而不用关心具体的API实现?不需要去了解Java和Scala的具体 ...
Apache Flink 零基础入门（三）编写最简单的helloWorld
实验环境 JDK 1.8 IDE Intellij idea Flink 1.8.1 实验内容创建一个Flink简单Demo,可以从流数据中统计单词个数. 实验步骤首先创建一个maven项目,其中 ...
Apache Flink 零基础入门（十五）Flink DataStream编程（如何自定义DataSource）
数据源可以通过StreamExecutionEnvironment.addSource(sourceFunction)方式来创建,Flink也提供了一些内置的数据源方便使用,例如readTextFil ...
Apache Flink 零基础入门（一）：基础概念解析
Apache Flink 的定义.架构及原理 Apache Flink 是一个分布式大数据处理引擎,可对有限数据流和无限数据流进行有状态或无状态的计算,能够部署在各种集群环境,对各种规模大小的数据进行 ...
Apache Flink 零基础入门（二十）Flink部署与作业的提交
之前我们都是基于Idea在本地进行开发,这种方式很适合开发以及测试,但是开发完之后,如何提交到服务器中运行? Flink单机部署方式本地开发和测试过程中非常有用,只要把代码放到服务器直接运行. 前置 ...
Apache Flink 零基础入门（二十）Flink kafka connector
内置source和sink 内置source包括从文件读取,从文件夹读取,从socket中读取.从集合或者迭代器中读取.内置的sink包括写文件.控制台输出.socket 内置connectors A ...
Apache Flink 零基础入门（十九）Flink windows和Time操作
Time类型在Flink中常用的Time类型: 处理时间摄取时间事件时间处理时间是上图中,最后一步的处理时间,表示服务器中执行相关操作的处理时间.例如一些算子操作时间,在服务器上面的时间. ...
Apache Flink 零基础入门（十四）Flink 分布式缓存
Apache Flink 提供了一个分布式缓存,类似于Hadoop,用户可以并行获取数据. 通过注册一个文件或者文件夹到本地或者远程HDFS等,在getExecutionEnvironment中指定一 ...

Apache Flink 零基础入门（十七）Flink 自定义Sink

Apache Flink 零基础入门（十七）Flink 自定义Sink相关推荐

最新文章

热门文章