DataStream 中的 DataSet Source 数据源

Author: Java-Edge

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

@@ -0,0 +1,27 @@
+package com.javaedge.java.chapter4;
+
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+
+/**
+ * @author JavaEdge
+ * @date 2019-07-20
+ */
+public class JavaCustomNonParallelSourceFunction implements SourceFunction<Long> {
+    boolean isRunning = true;
+    long count = 1;
+
+    @Override
+    public void run(SourceFunction.SourceContext<Long> ctx) throws Exception {
+        while (true) {
+            ctx.collect(count);
+            count += 1;
+            Thread.sleep(1000);
+        }
+    }
+
+    @Override
+    public void cancel() {
+        isRunning = false;
+    }
+}
+

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

@@ -0,0 +1,28 @@
+package com.javaedge.java.chapter4;
+
+import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+
+/**
+ * @author JavaEdge
+ * @date 2019-07-20
+ */
+public class JavaCustomParallelSourceFunction implements ParallelSourceFunction<Long> {
+    boolean isRunning = true;
+    long count = 1;
+
+    @Override
+    public void run(SourceFunction.SourceContext<Long> ctx) throws Exception {
+        while (true) {
+            ctx.collect(count);
+            count += 1;
+            Thread.sleep(1000);
+        }
+    }
+
+    @Override
+    public void cancel() {
+        isRunning = false;
+    }
+}
+

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

@@ -0,0 +1,37 @@
+package com.javaedge.java.chapter4;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
+
+/**
+ * @author JavaEdge
+ * @date 2019-07-20
+ */
+public class JavaCustomRichParallelSourceFunction extends RichParallelSourceFunction<Long> {
+    boolean isRunning = true;
+    long count = 1;
+
+    @Override
+    public void run(SourceContext<Long> ctx) throws Exception {
+        while (true) {
+            ctx.collect(count);
+            count += 1;
+            Thread.sleep(1000);
+        }
+    }
+
+    @Override
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+    }
+
+    @Override
+    public void close() throws Exception {
+        super.close();
+    }
+
+    @Override
+    public void cancel() {
+        isRunning = false;
+    }
+}

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

@@ -0,0 +1,46 @@
+package com.javaedge.java.chapter4;
+
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+
+/**
+ * @author JavaEdge
+ * @date 2019-07-19
+ */
+public class JavaDataStreamSourceApp {
+
+    public static void main(String[] args) throws Exception {
+        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+//      socketFunction(env);
+//      nonParallelSourceFunction(env);
+        parallelSourceFunction(env);
+
+        env.execute("JavaDataStreamSourceApp");
+    }
+
+    public static void nonParallelSourceFunction(StreamExecutionEnvironment env) {
+        DataStreamSource<Long> data = env.addSource(new JavaCustomNonParallelSourceFunction());
+        data.print().setParallelism(1);
+    }
+
+    public static void socketFunction(StreamExecutionEnvironment env) {
+        DataStreamSource<String> data = env.socketTextStream("localhost", 9999);
+        data.print().setParallelism(1);
+    }
+
+    public static void richParallelSourceFunction(StreamExecutionEnvironment env) {
+        DataStreamSource<Long> data = env.addSource(new JavaCustomRichParallelSourceFunction()).setParallelism(2);
+        data.print().setParallelism(1);
+    }
+
+
+    public static void parallelSourceFunction(StreamExecutionEnvironment env) {
+        DataStreamSource<Long> data = env.addSource(new JavaCustomParallelSourceFunction()).setParallelism(2);
+        data.print().setParallelism(1);
+    }
+
+
+//
+}
+

src/main/scala/com/javaedge/scala/chapter4/CustomNonParallelSourceFunction.scala

@@ -0,0 +1,30 @@
+package com.javaedge.scala.chapter4
+
+import org.apache.flink.streaming.api.functions.source.SourceFunction
+
+/**
+  * 自定义
+  *
+  * @author JavaEdge
+  * @date 2019-07-19
+  *
+  */
+class CustomNonParallelSourceFunction extends SourceFunction[Long]{
+
+  var count = 1L
+  var isRunning = true
+
+  override def cancel(): Unit = {
+    isRunning = false
+  }
+
+  override def run(ctx: SourceFunction.SourceContext[Long]): Unit = {
+    while(isRunning) {
+      ctx.collect(count)
+      count += 1
+      Thread.sleep(1000)
+    }
+  }
+}
+
+

src/main/scala/com/javaedge/scala/chapter4/CustomParallelSourceFunction.scala

@@ -0,0 +1,28 @@
+package com.javaedge.scala.chapter4
+
+import org.apache.flink.streaming.api.functions.source.{ParallelSourceFunction, SourceFunction}
+
+/**
+  * 实现 ParallelSourceFunction
+  *
+  * @author JavaEdge
+  * @date 2019-07-20
+  *
+  */
+class CustomParallelSourceFunction extends ParallelSourceFunction[Long]{
+
+  var isRunning = true
+  var count = 1L
+
+  override def cancel(): Unit = {
+    isRunning = false
+  }
+
+  override def run(ctx: SourceFunction.SourceContext[Long]): Unit = {
+    while(isRunning) {
+      ctx.collect(count)
+      count += 1
+      Thread.sleep(1000)
+    }
+  }
+}

src/main/scala/com/javaedge/scala/chapter4/CustomRichParallelSourceFunction.scala

@@ -0,0 +1,29 @@
+package com.javaedge.scala.chapter4
+
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
+
+/**
+  * @author JavaEdge
+  * @date 2019-07-20
+  */
+class CustomRichParallelSourceFunction extends RichParallelSourceFunction[Long] {
+  var isRunning = true
+  var count = 1L
+
+  override def cancel(): Unit = {
+    isRunning = false
+  }
+
+  override def open(parameters: Configuration): Unit = super.open(parameters)
+
+  override def close(): Unit = super.close()
+
+  override def run(ctx: SourceFunction.SourceContext[Long]): Unit = {
+    while (isRunning) {
+      ctx.collect(count)
+      count += 1
+      Thread.sleep(1000)
+    }
+  }
+}

src/main/scala/com/javaedge/scala/chapter4/DataStreamSourceApp.scala

@@ -0,0 +1,43 @@
+package com.javaedge.scala.chapter4
+
+import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
+import org.apache.flink.api.scala._
+
+/**
+  * @author JavaEdge
+  * @date 2019-07-19
+  */
+object DataStreamSourceApp {
+
+  def main(args: Array[String]): Unit = {
+    val env = StreamExecutionEnvironment.getExecutionEnvironment
+    // socketFunction(env)
+    nonParallelSourceFunction(env)
+    env.execute("DataStreamSourceApp")
+  }
+
+  def nonParallelSourceFunction(env: StreamExecutionEnvironment): Unit = {
+    val data = env.addSource(new CustomNonParallelSourceFunction)
+    data.print().setParallelism(1)
+  }
+
+  def socketFunction(env: StreamExecutionEnvironment): Unit = {
+
+    val data = env.socketTextStream("localhost", 9999)
+    data.print().setParallelism(1)
+  }
+
+  //  def richParallelSourceFunction(env:StreamExecutionEnvironment): Unit = {
+  //    val data = env.addSource(new CustomRichParallelSourceFunction).setParallelism(2)
+  //    data.print()
+  //  }
+  //
+  //  def parallelSourceFunction(env:StreamExecutionEnvironment): Unit = {
+  //    val data = env.addSource(new CustomParallelSourceFunction).setParallelism(2)
+  //    data.print()
+  //  }
+  //
+  //
+
+}
+