Flink Maven项目准备、key的定义、DataSet的数据源

虾米姐阅读：268 2022-06-06 14:18:37 评论：0

1. Maven项目准备

resources/log4j2.properties

################################################################################ 
#  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. 
################################################################################ 
 
rootLogger.level = INFO 
rootLogger.appenderRef.console.ref = ConsoleAppender 
 
appender.console.name = ConsoleAppender 
appender.console.type = CONSOLE 
appender.console.layout.type = PatternLayout 
appender.console.layout.pattern = %d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n

pom.xml

<?xml version="1.0" encoding="UTF-8"?> 
<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"> 
    <modelVersion>4.0.0</modelVersion> 
 
    <groupId>com.hh</groupId> 
    <artifactId>flink-test</artifactId> 
    <version>1.0-SNAPSHOT</version> 
 
    <properties> 
        <flink.version>1.13.2</flink.version> 
        <scala.binary.version>2.11</scala.binary.version> 
        <scala.version>2.11.12</scala.version> 
        <target.java.version>1.8</target.java.version> 
    </properties> 
 
    <repositories> 
        <repository> 
            <id>central</id> 
            <name>Central Repository</name> 
            <url>https://repo.maven.apache.org/maven2</url> 
            <layout>default</layout> 
            <snapshots> 
                <enabled>false</enabled> 
            </snapshots> 
        </repository> 
    </repositories> 
 
    <dependencies> 
 
        <dependency> 
            <groupId>org.apache.flink</groupId> 
            <artifactId>flink-scala_${scala.binary.version}</artifactId> 
            <version>${flink.version}</version> 
            <scope>provided</scope> 
        </dependency> 
 
        <dependency> 
            <groupId>org.apache.flink</groupId> 
            <artifactId>flink-clients_${scala.binary.version}</artifactId> 
            <version>${flink.version}</version> 
            <scope>provided</scope> 
        </dependency> 
 
        <dependency> 
            <groupId>org.scala-lang</groupId> 
            <artifactId>scala-library</artifactId> 
            <version>${scala.version}</version> 
            <scope>provided</scope> 
        </dependency> 
 
    </dependencies> 
 
    <build> 
        <plugins> 
 
            <plugin> 
                <groupId>org.apache.maven.plugins</groupId> 
                <artifactId>maven-shade-plugin</artifactId> 
                <version>3.2.4</version> 
                <executions> 
                    <execution> 
                        <phase>package</phase> 
                        <goals> 
                            <goal>shade</goal> 
                        </goals> 
                        <configuration> 
                            <artifactSet> 
                                <excludes> 
                                    <exclude></exclude> 
                                </excludes> 
                            </artifactSet> 
                            <filters> 
                                <filter> 
                                    <artifact>*:*</artifact> 
                                    <excludes> 
                                        <exclude>META-INF/*.RSA</exclude> 
                                    </excludes> 
                                </filter> 
                            </filters> 
                            <transformers> 
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> 
                                    <mainClass></mainClass> 
                                </transformer> 
                            </transformers> 
                        </configuration> 
                    </execution> 
                </executions> 
            </plugin> 
 
            <plugin> 
                <groupId>org.apache.maven.plugins</groupId> 
                <artifactId>maven-compiler-plugin</artifactId> 
                <version>3.8.1</version> 
                <configuration> 
                    <source>${target.java.version}</source> 
                    <target>${target.java.version}</target> 
                </configuration> 
            </plugin> 
 
            <plugin> 
                <groupId>net.alchim31.maven</groupId> 
                <artifactId>scala-maven-plugin</artifactId> 
                <version>4.5.3</version> 
                <executions> 
                    <execution> 
                        <goals> 
                            <goal>compile</goal> 
                            <goal>testCompile</goal> 
                        </goals> 
                    </execution> 
                </executions> 
                <configuration> 
                    <args> 
                        <arg>-nobootcp</arg> 
                        <arg>-target:jvm-${target.java.version}</arg> 
                    </args> 
                </configuration> 
            </plugin> 
 
        </plugins> 
    </build> 
 
</project>

2. 使用字段表达式为groupBy定义嵌套数据类型的Key

== keyBy同理==

package devBase 
 
import org.apache.flink.api.scala.{ExecutionEnvironment,createTypeInformation} 
 
case class Score(english:Double, math:Double) 
case class Teacher(name:String, student:(String, Score)) 
 
object DefineKey { 
 
 
  def main(args: Array[String]): Unit = { 
 
 
    val env = ExecutionEnvironment.getExecutionEnvironment 
    val input = env.fromElements(Teacher("teacher1",("student1", Score(81, 82))), 
      Teacher("teacher2",("student2", Score(91, 92)))) 
    input.groupBy("student._2.math") 
    // 对整个teacher进行分组；也可以用于普通class等其它数据类型 
    input.groupBy("_") 
 
  } 
 
}

3. DataSet API的数据源

3.1 基于文件

readTextFile.txt文件内容如下：

hello 
world

readCsvFile.csv文件内容如下：

"Liming",1,"Bei,jing" 
comment_Zhangsan,true,Shanghai 
Zhaosi,False,Guangzhou

测试代码如下：

package devBase 
 
import org.apache.flink.api.java.io.TextInputFormat 
import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment, createTypeInformation} 
import org.apache.flink.core.fs.Path 
import org.apache.flink.types.StringValue 
 
object DatasetApiTest { 
 
  def main(args: Array[String]): Unit = { 
 
    val env = ExecutionEnvironment.getExecutionEnvironment 
    val text_filepath = "src/main/resources/readTextFile.txt" 
    val text_input: DataSet[String] = env.readTextFile(text_filepath) 
    text_input.print() 
    /* 
    world 
    hello 
     */ 
 
    // StringValue为Flink定义的可变字符串 
    val stringValue_input: DataSet[StringValue] = env.readTextFileWithValue(text_filepath) 
    stringValue_input.print() 
    /* 
    world 
    hello 
     */ 
 
    // 读取一个原始数据类型，如String、Int 
    val primitives_input: DataSet[String] = env.readFileOfPrimitives[String](text_filepath, "\n") 
    primitives_input.print() 
    /* 
    hello 
    world 
     */ 
 
    val file_input = env.readFile(new TextInputFormat(new Path(text_filepath)), text_filepath) 
    file_input.print() 
    /* 
    hello 
    world 
     */ 
 
    val create_input = env.createInput(new TextInputFormat(new Path(text_filepath))) 
    create_input.print() 
    /* 
    world 
    hello 
     */ 
 
 
    val csv_input: DataSet[(Boolean, String)] = env.readCsvFile( 
      "src/main/resources/readCsvFile.csv",         // 读取的文件路径，该参数必须指定，其它参数可不指定 
      "\n",                   // 每行数据分隔符 
      ",",                    // 字段分隔符 
      Character.valueOf('"'),               // 字符串引号字符 
      false,                  // 是否忽略第一行数据 
      "comment_",           // 以该字符串开头的行数据，直接忽略 
      false,                        // true表示忽略解析错误的行，false遇到解析错误的行直接报错 
      Array(1, 2),                          // 需要从CSV文件获取的字段列表 
      Array("is_girl", "city")              // 给获取的字段列表定义列名 
    ) 
    csv_input.print() 
    /* 
    (true,Bei,jing) 
    (false,Guangzhou) 
     */ 
  } 
 
}

3.2 压缩文件

Flink的任何FileInputFormat(如TextInputFormat)都支持直接读取压缩文件，支持的压缩文件后缀有GZip(.gz、.gzip)、Bzip2(.bz2)、XZ(.xz)，下面以.gz为例进行演示

在linux准备.gz压缩文件

[root@bigdata005 opt]#  
[root@bigdata005 opt]# cat readZipFile.txt  
hello 
world[root@bigdata005 opt]#  
[root@bigdata005 opt]#  
[root@bigdata005 opt]# gzip readZipFile.txt 
[root@bigdata005 opt]#  
[root@bigdata005 opt]# ll readZipFile* 
-rw-r--r--. 1 root root 48 9月   6 22:02 readZipFile.txt.gz 
[root@bigdata005 opt]#

将readZipFile.txt.gz压缩文件复制到IDEA的src/main/resources目录下

完整读取代码如下：

package devBase 
 
import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment} 
 
object DatasetApiTest { 
 
  def main(args: Array[String]): Unit = { 
 
    val env = ExecutionEnvironment.getExecutionEnvironment 
 
    val zip_input:DataSet[String] = env.readTextFile("src/main/resources/readZipFile.txt.gz") 
    zip_input.print() 
  } 
 
}

执行结果如下：

hello 
world

3.3 基于集合

package devBase 
 
import org.apache.flink.api.scala.{ExecutionEnvironment, createTypeInformation} 
import org.apache.flink.util.NumberSequenceIterator 
 
import scala.collection.mutable.ArrayBuffer 
 
object DatasetApiTest { 
 
  def main(args: Array[String]): Unit = { 
 
    val env = ExecutionEnvironment.getExecutionEnvironment 
    val input1 = env.fromElements(("Liming", 10), ("Zhangsan", 20)) 
    input1.print() 
    /* 
    (Liming,10) 
    (Zhangsan,20) 
     */ 
 
    val input2 = env.fromCollection(ArrayBuffer(("Liming", 10), ("Zhangsan", 20))) 
    input2.print() 
    /* 
    (Liming,10) 
    (Zhangsan,20) 
     */ 
 
    // 参数为：SplittableIterator[T]， 本示例生成0，1，2，3的序列 
    val input3 = env.fromParallelCollection(new NumberSequenceIterator(0L, 3L)) 
    input3.print() 
    /* 
    3 
    1 
    2 
    0 
     */ 
 
    // 生成0，1，2，3的序列 
    val input4 = env.generateSequence(0L, 3L) 
    input4.print() 
    /* 
    3 
    2 
    0 
    1 
     */ 
  } 
 
}

3.4 zip DataSet中的元素

zipWithIndex

唯一ID是连续的，需要计算每个分区的数据量

package devBase 
 
import org.apache.flink.api.scala.utils.DataSetUtils 
import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment, createTypeInformation} 
 
object DatasetApiTest { 
 
  def main(args: Array[String]): Unit = { 
 
    val env = ExecutionEnvironment.getExecutionEnvironment 
 
    val input = env.fromElements("1A", "2B", "3C", "4D", "5E", "6F", "7G", "8H") 
      .setParallelism(4) 
 
    val zipWithIndex_output:DataSet[(Long,String)] = input.zipWithIndex 
    zipWithIndex_output.print() 
     
  } 
 
}

执行结果

(0,1A) 
(1,5E) 
(2,2B) 
(3,6F) 
(4,3C) 
(5,7G) 
(6,4D) 
(7,8H)

zipWithUniqueId

唯一ID是不连续的，如共5个ID，分别为(0、1、3、6、8)，不需要计算每个分区的数据量

package devBase 
 
import org.apache.flink.api.scala.utils.DataSetUtils 
import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment, createTypeInformation} 
 
object DatasetApiTest { 
 
  def main(args: Array[String]): Unit = { 
 
    val env = ExecutionEnvironment.getExecutionEnvironment 
 
    val input = env.fromElements("1A", "2B", "3C", "4D", "5E", "6F", "7G", "8H") 
      .setParallelism(4) 
 
    val zipWithUniqueId_output:DataSet[(Long,String)] = input.zipWithUniqueId 
    zipWithUniqueId_output.print() 
 
  } 
 
}

执行结果：

(0,1A) 
(4,5E) 
(1,2B) 
(5,6F) 
(2,3C) 
(6,7G) 
(3,4D) 
(7,8H)

标签:程序员

声明

1.本站遵循行业规范，任何转载的稿件都会明确标注作者和来源；2.本站的原创文章，请转载时务必注明文章作者和来源，不尊重原创的行为我们将追究责任；3.作者投稿可能会经我们编辑修改或补充。