9.FLINK SinkAPI自定义sink

Posted 2022-07-12 to.to

9.Sink
9.1.API
9.2.注意
9.3.自定义sink

9.Sink

9.1.API

1、ds.print 直接输出到控制台
2、ds.printToErr()直接输出到控制台，用红色
3、ds.writeAsText(“本地/HDFS的path”,WriteMode.OVERWRITE).setParallelism(1);

9.2.注意

在输出到path的时候，可以在前面设置并行度，如果
并行度>1,则path为目录
并行度=1,则path为文件名

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author tuzuoquan
 * @date 2022/4/19 22:46
 */
public class SinkDemo01 

    public static void main(String[] args) throws Exception 
        //TODO 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        //TODO 1.source
        DataStream<String> ds = env.readTextFile("D:/tmp/stderr.log");

        //TODO 2.transformation
        //TODO 3.sink
        ds.print();
        ds.print("输出标识");
        //会在控制台上以红色输出
        ds.printToErr();
        //会在控制台上以红色输出
        ds.printToErr("输出标识");
        ds.writeAsText("D:/tmp/output/result1").setParallelism(1);
        ds.writeAsText("D:/tmp/output/result1").setParallelism(2);

        //TODO 4.execute
        env.execute();

    

输出结果：
8> bbb ccc eee ttt wwww qqqq aaaa dddd xxxx
输出标识:8> bbb ccc eee ttt wwww qqqq aaaa dddd xxxx
3> eee ggg hhvvv vvvvddd dddd dddd ssss xxxx
输出标识:3> eee ggg hhvvv vvvvddd dddd dddd ssss xxxx
1> xxxxx cccc vvv bbbb nnnn mmmm mmmmm uuuu
输出标识:1> xxxxx cccc vvv bbbb nnnn mmmm mmmmm uuuu
6> sss rrrr vvv ffff hhh iii ooo ppp lll yyy
输出标识:6> sss rrrr vvv ffff hhh iii ooo ppp lll yyy
5> sss  bbb ccc ddd eee fff ddd xxx kkk nnn
输出标识:5> sss  bbb ccc ddd eee fff ddd xxx kkk nnn
8> bbb ccc eee ttt wwww qqqq aaaa dddd xxxx
输出标识:8> bbb ccc eee ttt wwww qqqq aaaa dddd xxxx
3> eee ggg hhvvv vvvvddd dddd dddd ssss xxxx
输出标识:3> eee ggg hhvvv vvvvddd dddd dddd ssss xxxx
1> xxxxx cccc vvv bbbb nnnn mmmm mmmmm uuuu
输出标识:1> xxxxx cccc vvv bbbb nnnn mmmm mmmmm uuuu
6> sss rrrr vvv ffff hhh iii ooo ppp lll yyy
输出标识:6> sss rrrr vvv ffff hhh iii ooo ppp lll yyy
5> sss  bbb ccc ddd eee fff ddd xxx kkk nnn
输出标识:5> sss  bbb ccc ddd eee fff ddd xxx kkk nnn

另外，在D:/tmp/output/result1和D:/tmp/output/result2中有数据。

9.3.自定义sink

需求：将FLink集合中的数据通过自定义Sink保存到mysql

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

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

/**
 * @author tuzuoquan
 * @date 2022/4/21 21:52
 */
public class SinkDemo02 

    public static void main(String[] args) throws Exception 
        //TODO 1.source
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        //TODO 1.source
        DataStream<Student> studentDS = env.fromElements(new Student(null, "tony", 18));
        //TODO 2.transformation
        //TODO 3.sink
        studentDS.addSink(new MySQLSink());

        //TODO 4.execute
        env.execute();
    

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Student 
        private Integer id;
        private String name;
        private Integer age;
    

    public static class MySQLSink extends RichSinkFunction<Student> 
        private Connection conn = null;
        private PreparedStatement ps = null;

        @Override
        public void open(Configuration parameters) throws Exception 
            conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata", "root", "root");
            String sql = "INSERT INTO `t_student` (`id`, `name`, `age`) VALUES (null, ?, ?);";
            ps = conn.prepareStatement(sql);
        

        @Override
        public void invoke(Student value, Context context) throws Exception 
            //设置?占位符参数值
            ps.setString(1, value.getName());
            ps.setInt(2, value.getAge());
            //执行sql
            ps.executeUpdate();
        

        @Override
        public void close() throws Exception 
            if (conn != null)  conn.close(); 
            if (ps != null)  ps.close();