DataStream 中的 算子及DataSet Sinks

Author: Java-Edge

.idea/dataSources.xml

@@ -82,6 +82,12 @@
 			<scope>runtime</scope>
 		</dependency>
 
+		<dependency>
+			<groupId>mysql</groupId>
+			<artifactId>mysql-connector-java</artifactId>
+			<version>8.0.13</version>
+		</dependency>
+
 		<dependency>
 			<groupId>org.projectlombok</groupId>
 			<artifactId>lombok</artifactId>

.idea/dictionaries/sss.xml

@@ -0,0 +1,33 @@
+package com.javaedge.java.chapter4;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+/**
+ * @author JavaEdge
+ *
+ * @date 2019-07-21
+ */
+public class JavaCustomSinkToMySQL {
+    public static void main(String[] args) throws Exception {
+        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+        DataStreamSource<String> source =  env.socketTextStream("localhost", 7777);
+
+        SingleOutputStreamOperator<Student> studentStream =  source.map(new MapFunction<String, Student>() {
+            @Override
+            public Student map(String value) throws Exception {
+                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;
+            }
+        });
+        studentStream.addSink(new SinkToMySQL());
+        env.execute("JavaCustomSinkToMySQL");
+    }
+}

.idea/inspectionProfiles/Project_Default.xml

@@ -0,0 +1,70 @@
+package com.javaedge.java.chapter4;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.streaming.api.collector.selector.OutputSelector;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.datastream.SplitStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author JavaEdge
+ * @date 2019-07-20
+ */
+public class JavaDataStreamTransformationApp {
+    public static void main(String[] args) throws Exception {
+        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+//      filterFunction(env);
+//      unionFunction(env);
+        splitSelectFunction(env);
+        env.execute("JavaDataStreamTransformationApp");
+    }
+
+    public static void splitSelectFunction(StreamExecutionEnvironment env) {
+        DataStreamSource<Long> data = env.addSource(new JavaCustomNonParallelSourceFunction());
+
+        SplitStream<Long> splits = data.split(new OutputSelector<Long>() {
+            @Override
+            public Iterable<String> select(Long value) {
+                List<String> output = new ArrayList<>();
+                if (value % 2 == 0) {
+                    output.add("even");
+                } else {
+                    output.add("odd");
+                }
+                return output;
+            }
+        });
+
+        splits.select("odd").print().setParallelism(1);
+    }
+
+    public static void unionFunction(StreamExecutionEnvironment env) {
+        DataStreamSource<Long> data1 = env.addSource(new JavaCustomNonParallelSourceFunction());
+        DataStreamSource<Long> data2 = env.addSource(new JavaCustomNonParallelSourceFunction());
+        data1.union(data2).print().setParallelism(1);
+    }
+
+    public static void filterFunction(StreamExecutionEnvironment env) {
+        DataStreamSource<Long> data = env.addSource(new JavaCustomNonParallelSourceFunction());
+        data.map(new MapFunction<Long, Long>() {
+            @Override
+            public Long map(Long value) throws Exception {
+                System.out.println("value = [" + value + "]");
+                return value;
+            }
+        }).filter(new FilterFunction<Long>() {
+            @Override
+            public boolean filter(Long value) throws Exception {
+                return value % 2 == 0;
+            }
+        }).print().setParallelism(1);
+    }
+
+
+}
+
+

.idea/misc.xml

@@ -0,0 +1,91 @@
+package com.javaedge.java.chapter4;
+
+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;
+
+/**
+ * @author JavaEdge
+ *
+ * @date 2019-07-21
+ */
+public class SinkToMySQL 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/javaedge_flink";
+
+            conn = DriverManager.getConnection(url,"root","root");
+
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+
+        return conn;
+    }
+
+    /**
+     * 在open方法中建立connection
+     * @param parameters
+     * @throws Exception
+     */
+    @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");
+
+    }
+
+    /**
+     * 每条记录插入时调用一次
+     *
+     * @param value
+     * @param context
+     * @throws Exception
+     */
+    @Override
+    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();
+
+    }
+
+    /**
+     * 在close方法中要释放资源
+     * @throws Exception
+     */
+    @Override
+    public void close() throws Exception {
+        super.close();
+
+        if(pstmt != null) {
+            pstmt.close();
+        }
+
+        if(connection != null) {
+            connection.close();
+        }
+    }
+}
+

pom.xml

@@ -0,0 +1,18 @@
+package com.javaedge.java.chapter4;
+
+import lombok.Data;
+
+/**
+ * @author JavaEdge
+ * @date 2019-07-21
+ */
+@Data
+public class Student {
+
+    private int id;
+
+    private String name;
+
+    private int age;
+}
+

src/main/java/com/javaedge/java/chapter4/JavaCustomSinkToMySQL.java

@@ -0,0 +1,57 @@
+package com.javaedge.scala.chapter4
+
+import java.{lang, util}
+
+import org.apache.flink.streaming.api.collector.selector.OutputSelector
+import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
+
+/**
+  * @author JavaEdge
+  * @date 2019-07-20
+  */
+object DataStreamTransformationApp {
+
+  def main(args: Array[String]): Unit = {
+    val env = StreamExecutionEnvironment.getExecutionEnvironment
+    //    filterFunction(env)
+    unionFunction(env)
+    env.execute("DataStreamTransformationApp")
+  }
+
+  def splitSelectFunction(env: StreamExecutionEnvironment): Unit = {
+    import org.apache.flink.api.scala._
+    val data = env.addSource(new CustomNonParallelSourceFunction)
+
+    val splits = data.split(new OutputSelector[Long] {
+      override def select(value: Long): lang.Iterable[String] = {
+        val list = new util.ArrayList[String]()
+        if (value % 2 == 0) {
+          list.add("even")
+        } else {
+          list.add("odd")
+        }
+        list
+      }
+    })
+    splits.select("even", "odd").print().setParallelism(1)
+  }
+
+  def unionFunction(env: StreamExecutionEnvironment): Unit = {
+    import org.apache.flink.api.scala._
+    val data1 = env.addSource(new CustomNonParallelSourceFunction)
+    val data2 = env.addSource(new CustomNonParallelSourceFunction)
+    data1.union(data2).print().setParallelism(1)
+  }
+
+  def filterFunction(env: StreamExecutionEnvironment): Unit = {
+    import org.apache.flink.api.scala._
+    val data = env.addSource(new CustomNonParallelSourceFunction)
+
+    data.map(x => {
+      println("received: " + x)
+      x
+    }).filter(_ % 2 == 0).print().setParallelism(1)
+  }
+
+
+}