DataSet Sink

Author: Java-Edge

src/main/java/com/javaedge/java/chapter3/DataSetTransformationApp.java

@@ -36,7 +36,7 @@ public static void crossFunction(ExecutionEnvironment env) throws Exception {
         List<String> info1 = new ArrayList<>();
         info1.add("曼联");
         info1.add("曼城");
-        
+
         List<String> info2 = new ArrayList<>();
         info2.add("3");
         info2.add("1");

src/main/java/com/javaedge/java/chapter3/JavaDataSetSinkApp.java

@@ -0,0 +1,33 @@
+package com.javaedge.java.chapter3;
+
+/**
+ * @author sss
+ * @date 2019-07-18
+ */
+
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.DataSource;
+import org.apache.flink.core.fs.FileSystem;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author JavaEdge
+ */
+public class JavaDataSetSinkApp {
+
+    public static void main(String[] args) throws Exception {
+
+        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+        List<Integer> info = new ArrayList<>();
+        for (int i = 1; i <= 10; i++) {
+            info.add(i);
+        }
+
+        String filePath = "/Volumes/doc/data/cp3/sink-out/";
+        DataSource<Integer> data = env.fromCollection(info);
+        data.writeAsText(filePath, FileSystem.WriteMode.OVERWRITE);
+        env.execute("JavaDataSetSinkApp");
+    }
+}

src/main/scala/com/javaedge/scala/chapter3/DataSetSinkApp.scala

@@ -0,0 +1,22 @@
+package com.javaedge.scala.chapter3
+
+import org.apache.flink.api.scala.ExecutionEnvironment
+import org.apache.flink.core.fs.FileSystem.WriteMode
+import org.apache.flink.api.scala._
+/**
+  * @author JavaEdge
+  *
+  * @date 2019-07-18
+  */
+object DataSetSinkApp {
+  def main(args: Array[String]): Unit = {
+
+    val env = ExecutionEnvironment.getExecutionEnvironment
+    val data = 1.to(10)
+    val text = env.fromCollection(data)
+
+    val filePath = "/Volumes/doc/data/cp3/sink-out/"
+    text.writeAsText(filePath, WriteMode.OVERWRITE).setParallelism(5)
+    env.execute("DataSetSinkApp")
+  }
+}