目录
1. Maven项目准备
- 1. Maven项目准备
- 2. 使用字段表达式为groupBy定义嵌套数据类型的Key
- 3. DataSet API的数据源
- 3.1 基于文件
- 3.2 压缩文件
- 3.3 基于集合
- 3.4 zip DataSet中的元素
- 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
4.0.0
com.hh
flink-test
1.0-SNAPSHOT
1.13.2
2.11
2.11.12
1.8
central
Central Repository
https://repo.maven.apache.org/maven2
default
false
org.apache.flink
flink-scala_${scala.binary.version}
${flink.version}
provided
org.apache.flink
flink-clients_${scala.binary.version}
${flink.version}
provided
org.scala-lang
scala-library
${scala.version}
provided
org.apache.maven.plugins
maven-shade-plugin
3.2.4
package
shade
*:*
META-INF/*.RSA
org.apache.maven.plugins
maven-compiler-plugin
3.8.1
${target.java.version}
${target.java.version}
net.alchim31.maven
scala-maven-plugin
4.5.3
compile
testCompile
-nobootcp
-target:jvm-${target.java.version}
== 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("_")
}
}
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)
*/
}
}
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
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
*/
}
}
- 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)
