transformation算子

Author: Java-Edge

.idea/misc.xml

@@ -1,4 +1,4 @@
-# 0 Flink 从入门到实战 - Scala版本
+# Flink 从入门到实战 - Scala/Java双语言版本
 
 # 1 核心知识点
 ![](https://upload-images.jianshu.io/upload_images/16782311-dc4156dc0a34d557.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
@@ -79,19 +79,11 @@ Flink常见的监控方式
 
 常见的Flink调优策略
 
-# 2 实战项目：基于Flink的互联网直播平台日志分析
-#### 互联网直播平台日志分析（以Scala语言开发）
-![](https://upload-images.jianshu.io/upload_images/16782311-ff3bd3a8d89089ba.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
-
-# 3 以下知名公司都在使用Flink
+# 2 以下知名公司都在使用Flink
 ![](https://upload-images.jianshu.io/upload_images/16782311-9447a7c96178832b.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
 
 # X 交流学习
 ![](https://upload-images.jianshu.io/upload_images/16782311-8d7acde57fdce062.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
-
 ## [Java交流群](https://jq.qq.com/?_wv=1027&k=5UB4P1T)
-![](https://upload-images.jianshu.io/upload_images/16782311-11d3533436ffbbdf.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
-
 ## [博客](http://www.shishusheng.com)
-
 ## [Github](https://github.com/Wasabi1234)

README.md

@@ -1,21 +1,3 @@
-<!--
-Licensed to the Apache Software Foundation (ASF) under one
-or more contributor license agreements.  See the NOTICE file
-distributed with this work for additional information
-regarding copyright ownership.  The ASF licenses this file
-to you under the Apache License, Version 2.0 (the
-"License"); you may not use this file except in compliance
-with the License.  You may obtain a copy of the License at
-
-  http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing,
-software distributed under the License is distributed on an
-"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-KIND, either express or implied.  See the License for the
-specific language governing permissions and limitations
-under the License.
--->
 <project xmlns="http://maven.apache.org/POM/4.0.0"
          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
@@ -99,6 +81,12 @@ under the License.
 			<version>1.2.17</version>
 			<scope>runtime</scope>
 		</dependency>
+
+		<dependency>
+			<groupId>org.projectlombok</groupId>
+			<artifactId>lombok</artifactId>
+			<version>1.18.8</version>
+		</dependency>
 	</dependencies>
 
 	<build>

pom.xml

@@ -0,0 +1,11 @@
+/**
+ * @author JavaEdge
+ *
+ * @date 2019-07-16
+ */
+public class TestJava {
+
+    public static void main(String[] args) {
+        System.out.println("Hello Java!");
+    }
+}

src/main/java/TestJava.java

@@ -0,0 +1,29 @@
+package com.javaedge.java.chapter3;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.DataSource;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author JavaEdge
+ * @date 2019-07-17
+ */
+public class DataSetTransformationApp {
+
+    public static void main(String[] args) throws Exception {
+        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+        mapFunction(env);
+    }
+
+    public static void mapFunction(ExecutionEnvironment env) throws Exception {
+        List<Integer> list = new ArrayList<>(10);
+        for (int i = 0; i < 10; i++) {
+            list.add(i);
+        }
+        DataSource<Integer> data = env.fromCollection(list);
+        data.map((MapFunction<Integer, Integer>) input -> input + 1).print();
+    }
+}

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

@@ -0,0 +1,36 @@
+package com.javaedge.java.chapter3;
+
+import org.apache.flink.api.java.ExecutionEnvironment;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author JavaEdge
+ *
+ * @date 2019-07-16
+ */
+public class JavaDataSetDataSourceApp {
+
+    public static void main(String[] args) throws Exception {
+        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+//        fromCollection(env);
+        textFile(env);
+    }
+
+    public static void textFile(ExecutionEnvironment env) throws Exception {
+        String filePath = "file:///Volumes/doc/data/data.txt";
+        env.readTextFile(filePath).print();
+        System.out.println("===========~这是一个分割线~============");
+
+        filePath = "file:///Volumes/doc/data/inputs";
+        env.readTextFile(filePath).print();
+    }
+    public static void fromCollection(ExecutionEnvironment env) throws Exception {
+        List<Integer> list = new ArrayList<>(10);
+        for (int i = 0; i < 10; i++) {
+            list.add(i);
+        }
+        env.fromCollection(list).print();
+    }
+}

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

@@ -0,0 +1,17 @@
+package com.javaedge.java.chapter3;
+
+import lombok.Data;
+import lombok.ToString;
+
+/**
+ * @author JavaEdge
+ *
+ * @date 2019-07-17
+ */
+@Data
+@ToString
+public class Person {
+    private String name;
+    private int age;
+    private String job;
+}

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

@@ -0,0 +1,41 @@
+package com.javaedge.scala.chapter3
+
+import org.apache.flink.api.scala.ExecutionEnvironment
+import org.apache.flink.api.scala._
+
+/**
+  * @author JavaEdge
+  * @date 2019-07-17
+  *
+  */
+object DataSetTransformationApp {
+
+  def main(args: Array[String]): Unit = {
+    val env = ExecutionEnvironment.getExecutionEnvironment
+    //    mapFunction(env)
+    filterFunction(env)
+  }
+
+  def filterFunction(env: ExecutionEnvironment): Unit = {
+    //    val data = env.fromCollection(List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
+    //    data.map(_ + 1).filter(_ > 5).print()
+
+    env.fromCollection(List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
+      .map(_ + 1)
+      .filter(_ > 5)
+      .print()
+  }
+
+  def mapFunction(env: ExecutionEnvironment): Unit = {
+    import org.apache.flink.api.scala._
+    val data = env.fromCollection(List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
+
+    // data.print()
+    // 对data中的元素都做加一处理
+    // data.map((x: Int) => x + 1).print()
+    // data.map((x) => x + 1).print()
+    // data.map(x => x + 1).print()
+    // 终极写法
+    data.map(_ + 1).print()
+  }
+}

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

@@ -0,0 +1,60 @@
+package com.javaedge.scala.chapter3
+
+import com.javaedge.java.chapter3.Person
+import org.apache.flink.api.scala.ExecutionEnvironment
+import org.apache.flink.configuration.Configuration
+
+/**
+  * @author JavaEdge
+  * @date 2019-07-16
+  */
+object ScalaDataSetDataSourceApp {
+  def main(args: Array[String]): Unit = {
+    val env = ExecutionEnvironment.getExecutionEnvironment
+    //    fromCollection(env)
+    //    textFile(env)
+    //    csvFile(env)
+    readCompressionFiles(env)
+  }
+
+  def readCompressionFiles(env:ExecutionEnvironment): Unit ={
+    val filePath = "file:///Volumes/doc/data/compress"
+    env.readTextFile(filePath).print()
+  }
+
+  def readRecursiveFiles(env: ExecutionEnvironment): Unit = {
+    // ❌
+    val filePath = "file:///Volumes/doc/data/nested"
+    env.readTextFile(filePath).print()
+    println("===========~这是一个分割线~============")
+
+    // ✔️
+    val parameters = new Configuration
+    parameters.setBoolean("recursive.file.enumeration", true)
+    env.readTextFile(filePath).withParameters(parameters).print()
+  }
+
+  def csvFile(env: ExecutionEnvironment): Unit = {
+    import org.apache.flink.api.scala._
+    val filePath = "file:///Volumes/doc/data/data.csv"
+    //    env.readCsvFile[(String, Int, String)](filePath, ignoreFirstLine = true).print()
+    //    env.readCsvFile[(String, Int)](filePath, ignoreFirstLine = true,includedFields = Array(0, 1)).print()
+
+    /*无法正常运行
+      case class MyCaseClass(name: String, age: Int)
+      env.readCsvFile[MyCaseClass](filePath,ignoreFirstLine = true,includedFields = Array(0, 1)).print()
+    */
+    env.readCsvFile[Person](filePath, ignoreFirstLine = true, pojoFields = Array("name", "age", "job")).print()
+  }
+
+  def textFile(env: ExecutionEnvironment): Unit = {
+    val filePath = "file:///Volumes/doc/data/inputs"
+    env.readTextFile(filePath).print()
+  }
+
+  def fromCollection(env: ExecutionEnvironment): Unit = {
+    import org.apache.flink.api.scala._
+    val data = 1 to 10
+    env.fromCollection(data).print()
+  }
+}

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

@@ -0,0 +1,13 @@
+/**
+  * @author JavaEdge
+  *
+  * @date 2019-07-16
+  *
+  */
+object TestScala {
+
+  def main(args: Array[String]): Unit = {
+    println("Hello Scala!")
+  }
+
+}