目录

Sink

预定义Sink

基于控制台和文件的Sink

自定义Sink

MySQL

Sink

预定义Sink

基于控制台和文件的Sink

• API

1.ds.print 直接输出到控制台

2.ds.printToErr() 直接输出到控制台,用红色

3.ds.writeAsText("本地/HDFS的path",WriteMode.OVERWRITE).setParallelism(1)

• 注意:

在输出到path的时候,可以在前面设置并行度,如果

并行度>1,则path为目录

并行度=1,则path为文件名

• 代码演示:

      package cn.itcast.sink;
      import org.apache.flink.core.fs.FileSystem;
      import org.apache.flink.streaming.api.datastream.DataStream;
      import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
      /**
       * Author itcast
       * Desc
       * 1.ds.print 直接输出到控制台
       * 2.ds.printToErr() 直接输出到控制台,用红色
       * 3.ds.collect 将分布式数据收集为本地集合
       * 4.ds.setParallelism(1).writeAsText("本地/HDFS的path",WriteMode.OVERWRITE)
       */
      public class SinkDemo01 {
          public static void main(String[] args) throws Exception {
              //1.env
              StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
              //2.source
              //DataStream<String> ds = env.fromElements("hadoop", "flink");
              DataStream<String> ds = env.readTextFile("data/input/words.txt");
              //3.transformation
              //4.sink
              ds.print();
              ds.printToErr();
              ds.writeAsText("data/output/test", FileSystem.WriteMode.OVERWRITE).setParallelism(2);
              //注意:
              //Parallelism=1为文件
              //Parallelism>1为文件夹
              //5.execute
              env.execute();
          }
      }
  
 

自定义Sink

MySQL

• 需求:

将Flink集合中的数据通过自定义Sink保存到MySQL

• 代码实现:

      package cn.itcast.sink;
      import lombok.AllArgsConstructor;
      import lombok.Data;
      import lombok.NoArgsConstructor;
      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 itcast
       * Desc
       * 使用自定义sink将数据保存到MySQL
       */
      public class SinkDemo02_MySQL {
          public static void main(String[] args) throws Exception {
              //1.env
              StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
              //2.Source
              DataStream<Student> studentDS = env.fromElements(new Student(null, "tonyma", 18));
              //3.Transformation
              //4.Sink
              studentDS.addSink(new MySQLSink());
              //5.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 {
                  //加载驱动,开启连接
                  //Class.forName("com.mysql.jdbc.Driver");
                  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中的?设置具体值
                  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();
              }
          }
      }
  
 

文章来源: lansonli.blog.csdn.net，作者：Lansonli，版权归原作者所有，如需转载，请联系作者。

原文链接：lansonli.blog.csdn.net/article/details/116223019