大数据Hadoop之——Flink DataStream API 和 DataSet API

大数据Hadoop之——Flink DataStream API 和 DataSet API

@[TOC]

一、DataStream API概述

Flink 中的 DataStream 程序是对数据流（例如过滤、更新状态、定义窗口、聚合）进行转换的常规程序。数据流的起始是从各种源（例如消息队列、套接字流、文件）创建的。结果通过 sink 返回，例如可以将数据写入文件或标准输出（例如命令行终端）。Flink 程序可以在各种上下文中运行，可以独立运行，也可以嵌入到其它程序中。任务执行可以运行在本地 JVM 中，也可以运行在多台机器的集群上。

二、什么是DataStream ？

DataStream API 得名于特殊的 DataStream 类，该类用于表示 Flink 程序中的数据集合。你可以认为 它们是可以包含重复项的不可变数据集合。这些数据可以是有界（有限）的，也可以是无界（无限）的，但用于处理它们的API是相同的。 DataStream 在用法上类似于常规的 Java 集合，但在某些关键方面却大不相同。它们是不可变的，这意味着一旦它们被创建，你就不能添加或删除元素。你也不能简单地察看内部元素，而只能使用 DataStream API 操作来处理它们，DataStream API 操作也叫作转换（transformation）。 你可以通过在 Flink 程序中添加 source 创建一个初始的 DataStream。然后，你可以基于 DataStream 派生新的流，并使用 map、filter 等 API 方法把 DataStream 和派生的流连接在一起。

三、DataStream 数据处理过程

1）Data Sources（数据源）

1、Data Sources 原理

官方文档一个数据 source 包括三个核心组件：分片（Splits）、分片枚举器（SplitEnumerator） 以及 源阅读器（SourceReader）。

分片（Split） 是对一部分 source 数据的包装，如一个文件或者日志分区。分片是 source 进行任务分配和数据并行读取的基本粒度。 源阅读器（SourceReader） 会请求分片并进行处理，例如读取分片所表示的文件或日志分区。SourceReader 在 TaskManagers 上的 SourceOperators 并行运行，并产生并行的事件流/记录流。 分片枚举器（SplitEnumerator） 会生成分片并将它们分配给 SourceReader。该组件在 JobManager 上以单并行度运行，负责对未分配的分片进行维护，并以均衡的方式将其分配给 reader。SplitEnumerator 被认为是整个 Source 的“大脑”。

2、Data Sources 实现方式

1）基于文件

Source 是你的程序从中读取其输入的地方。你可以用 StreamExecutionEnvironment.addSource(sourceFunction)将一个 source 关联到你的程序。Flink 自带了许多预先实现的 source functions，不过你仍然可以通过实现 SourceFunction 接口编写自定义的非并行 source，也可以通过实现 ParallelSourceFunction 接口或者继承 RichParallelSourceFunction 类编写自定义的并行 sources。 通过 StreamExecutionEnvironment 可以访问多种预定义的 stream source，source 连接器，请查看连接器文档。

readTextFile(path)：读取文本文件。 readFile(fileInputFormat, path) - 按照指定的文件输入格式读取（一次）文件。 readFile(fileInputFormat, path, watchType, interval, pathFilter, typeInfo) ：这是前两个方法内部调用的方法。它基于给定的 fileInputFormat 读取路径 path 上的文件。根据提供的 watchType 的不同，source 可能定期（每 interval 毫秒）监控路径上的新数据（watchType 为 FileProcessingMode.PROCESS_CONTINUOUSLY），或者处理一次当前路径中的数据然后退出（watchType 为 FileProcessingMode.PROCESS_ONCE)。使用 pathFilter，用户可以进一步排除正在处理的文件。

2）基于套接字

socketTextStream：套接字读取。元素可以由分隔符分隔。 3）基于集合 fromCollection(Collection) ：从 Java Java.util.Collection 创建数据流。集合中的所有元素必须属于同一类型。 fromCollection(Iterator, Class) ：从迭代器创建数据流。class 参数指定迭代器返回元素的数据类型。 fromElements(T ...) ：从给定的对象序列中创建数据流。所有的对象必须属于同一类型。 fromParallelCollection(SplittableIterator, Class) ：从迭代器并行创建数据流。class 参数指定迭代器返回元素的数据类型。 generateSequence(from, to) ：基于给定间隔内的数字序列并行生成数据流。

4）自定义

addSource：关联一个新的 source function。例如，你可以使用 addSource(new FlinkKafkaConsumer<>(...)) 来从 Apache Kafka 获取数据。更多详细信息见连接器。 2）DataStream Transformations（数据流转换//处理/算子） 【温馨提示】是用户通过算子能将一个或多个 DataStream 转换成新的 DataStream，在应用程序中可以将多个数据转换算子合并成一个复杂的数据流拓扑。这部分内容将描述 Flink DataStream API 中基本的数据转换API，数据转换后各种数据分区方式，以及算子的链接策略。

1、数据流转换

2、物理分区

Flink 也提供以下方法让用户根据需要在数据转换完成后对数据分区进行更细粒度的配置。

3、算子链和资源组

将两个算子链接在一起能使得它们在同一个线程中执行，从而提升性能。Flink 默认会将能链接的算子尽可能地进行链接(例如， 两个 map 转换操作)。此外， Flink 还提供了对链接更细粒度控制的 API 以满足更多需求。 如果想对整个作业禁用算子链，可以调用 StreamExecutionEnvironment.disableOperatorChaining()。下列方法还提供了更细粒度的控制。需要注 意的是， 这些方法只能在 DataStream 转换操作后才能被调用，因为它们只对前一次数据转换生效。例如，可以 someStream.map(...).startNewChain() 这样调用，而不能 someStream.startNewChain()这样。

3）Data Sinks（数据输出）

sink 连接器，请查看连接器文档。

Data sinks 使用 DataStream 并将它们转发到文件、套接字、外部系统或打印它们。Flink 自带了多种内置的输出格式，这些格式相关的实现封装在 DataStreams 的算子里：

writeAsText() / TextOutputFormat : 将元素按行写成字符串。通过调用每个元素的 toString() 方法获得字符串。 writeAsCsv(...) / CsvOutputFormat ：将元组写成逗号分隔值文件。行和字段的分隔符是可配置的。每个字段的值来自对象的 toString() 方法。 print() / printToErr()：在标准输出/标准错误流上打印每个元素的 toString() 值。 可选地，可以提供一个前缀（msg）附加到输出。这有助于区分不同的 print 调用。如果并行度大于1，输出结果将附带输出任务标识符的前缀。 writeUsingOutputFormat() / FileOutputFormat ：自定义文件输出的方法和基类。支持自定义 object 到 byte 的转换。 writeToSocket ：根据 SerializationSchema 将元素写入套接字。 addSink ： 调用自定义 sink function。Flink 捆绑了连接到其他系统（例如 Apache Kafka）的连接器，这些连接器被实现为 sink functions。

【温馨提示】DataStream 的 write*() 方法主要用于调试目的。它们不参与 Flink 的 checkpointing，这意味着这些函数通常具有至少有一次语义。刷新到目标系统的数据取决于 OutputFormat 的实现。这意味着并非所有发送到 OutputFormat 的元素都会立即显示在目标系统中。此外，在失败的情况下，这些记录可能会丢失。 为了将流可靠地、精准一次地传输到文件系统中，请使用 StreamingFileSink。此外，通过 .addSink(...) 方法调用的自定义实现也可以参与 Flink 的 checkpointing，以实现精准一次的语义。

旁路输出（分流）

旁路输出在Flink中叫作SideOutput，用途类似于DataStream#split，本质上是一个数据流的切分行为，按照条件将DataStream切分为多个子数据流，子数据流叫作旁路输出数据流，每个旁路输出数据流可以有自己的下游处理逻辑。

使用旁路输出时，首先需要定义用于标识旁路输出流的 OutputTag：

val outputTag = OutputTag[String]("side-output")

可以通过以下方法将数据发送到旁路输出：

ProcessFunction KeyedProcessFunction CoProcessFunction KeyedCoProcessFunction ProcessWindowFunction ProcessAllWindowFunction

【示例】

package com import org.apache.flink.streaming.api.functions.ProcessFunction import org.apache.flink.streaming.api.scala._ import org.apache.flink.util.Collector object myOutputTag { def main(args: Array[String]): Unit = { val env = StreamExecutionEnvironment.getExecutionEnvironment val input: DataStream[String] = env.readTextFile("flink/data/hello.txt") val outputTag = OutputTag[String]("side-output") val mainDataStream = input .process(new ProcessFunction[String, String] { override def processElement( value: String, ctx: ProcessFunction[String, String]#Context, out: Collector[String]): Unit = { // 发送数据到主要的输出 out.collect(value) // 发送数据到旁路输出 ctx.output(outputTag, "sideout-" + value) } }) // 获取outputTag并输出 mainDataStream.getSideOutput(outputTag).print() // 必须调用execute或者executeAsync()，下面会讲 env.execute("test OutputTag") } }

【问题】Caused by: java.lang.ClassNotFoundException: org.apache.commons.compress.compressors.zstandard.ZstdCompressorInputStream

【解决】在pom.xml添加下面依赖

org.apache.commons commons-compress 1.21

2）Flink 程序剖析（scala）

Flink 程序看起来像一个转换 DataStream 的常规程序。每个程序由相同的基本部分组成：

获取一个执行环境（execution environment）； 加载/创建初始数据； 指定数据相关的转换； 指定计算结果的存储位置； 触发程序执行。

1、 获取一个执行环境（execution environment）

val env = StreamExecutionEnvironment.getExecutionEnvironment

2、加载/创建初始数据

为了指定 data sources，执行环境提供了一些方法，支持使用各种方法从文件中读取数据：你可以直接逐行读取数据，像读 CSV 文件一样，或使用任何第三方提供的 source。下面是将一个文本文件作为一个行的序列来读。 val env = StreamExecutionEnvironment.getExecutionEnvironment

// 加载数据源val input: DataStream[String] = env.readTextFile("file:///path/to/file")

#### 3、指定数据相关的转换 ```bash val env = StreamExecutionEnvironment.getExecutionEnvironment val input: DataStream[String] = env.readTextFile("file:///path/to/file") // 例如一个 map 的转换如下： val mapped = input.map { x => x.toInt }

4、指定计算结果的存储位置

一旦你有了包含最终结果的 DataStream，你就可以通过创建 sink 把它写到外部系统。下面是一些用于创建 sink 的示例方法：

val env = StreamExecutionEnvironment.getExecutionEnvironment val input: DataStream[String] = env.readTextFile("flink/data/source") // 例如一个 map 的转换如下： val mapped = input.map { x => x.toInt } // 存储到文件，当然还可以执行更多的sink // writeAsText第二个参数来定义输出模式，它有以下两个可选值： // WriteMode.NO_OVERWRITE：当指定路径上不存在任何文件时，才执行写出操作； // WriteMode.OVERWRITE：不论指定路径上是否存在文件，都执行写出操作；如果原来已有文件，则进行覆盖。 mapped.writeAsText("flink/data/sink", FileSystem.WriteMode.OVERWRITE)

5、触发程序执行

一旦指定了完整的程序，需要调用 StreamExecutionEnvironment 的 execute()方法来触发程序执行。根据 ExecutionEnvironment 的类型，执行会在你的本地机器上触发，或将你的程序提交到某个集群上执行。execute() 方法将等待作业完成，然后返回一个 JobExecutionResult，其中包含执行时间和累加器结果。 如果不想等待作业完成，可以通过调用 StreamExecutionEnvironment 的 executeAsync() 方法来触发作业异步执行。它会返回一个 JobClient，你可以通过它与刚刚提交的作业进行通信。如下是使用 executeAsync() 实现 execute() 语义的示例。 final JobClient jobClient = env.executeAsync();

final JobExecutionResult jobExecutionResult = jobClient.getJobExecutionResult().get();

完整示例程序（[官网示例](https://nightlies.apache.org/flink/flink-docs-release-1.14/zh/docs/dev/datastream/overview/)） **【问题一】** > 【温馨提示】如果出现这种报错，一般就是IDEA 对scope为**provided**，这是IDEA的bug：`Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/flink/api/scala/typeutils/CaseClassTypeInfo` 【解决】 - 【第一种方式】把依赖范围调大或者直接去掉都行，不清楚的可以看我之前的[Java-Maven详解]，但是记住在打包的时候得加上。 - 【第二种方式】Run->Edit Configurations，设置如下：  **【问题二】** > 【问题】`Caused by: java.util.concurrent.CompletionException: java.lang.NoSuchMethodError: org.apache.commons.math3.stat.descriptive.rank.Percentile.withNaNStrategy(Lorg/apache/commons/math3/stat/ranking/NaNStrategy;)Lorg/apache/commons/math3/stat/descriptive/rank/Percentile;`hadoop-common中的commons-math3冲突导致。 【解决】排除hadoop-common中的commons-math3，设置如此： ```xml org.apache.hadoop hadoop-common 3.3.1 provided org.apache.commons commons-math3

先启动服务

$ nc -lk 9999

WindowWordCount源码如下：

import org.apache.flink.streaming.api.scala._ import org.apache.flink.streaming.api.windowing.time.Time object WindowWordCount { def main(args: Array[String]) { val env = StreamExecutionEnvironment.getExecutionEnvironment val text = env.socketTextStream("localhost", 9999) val counts = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } } .map { (_, 1) } .keyBy(_._1) .window(TumblingProcessingTimeWindows.of(Time.seconds(5))) .sum(1) counts.print() env.execute("Window Stream WordCount") } }

四、什么是DataSet？

Flink用DataStream 表示无界数据集，用DataSet表示有界数据集，前者用于流处理应用程序，后者用于批处理应用程序。从操作形式上看，DataStream 和 DataSet 与集合 Collection 有些相似，但两者有着本质的区别：

DataStream 和 DataSet 是不可变的数据集合，因此不可以想操作集合那样增加或者删除 DataStream 和 DataSet 中的元素，也不可以通过诸如下标等方式访问某个元素。 Flink 应用程序通过 Source 创建 DataStream 对象和 DataSet 对象，通过转换操作产生新的 DataStream 对象和 DataSet 对象。 运行时是应用程序被调度执行时的上下文环境，通过StreamExecutionEnvironment或ExecutionEnvironment方法会根据当前环境自动选择本地或者集群运行时环境。

五、DataSet 数据处理过程

1）Data Sources （数据源）

数据源创建初始数据集，比如从文件或Java集合创建数据集。创建数据集的一般机制抽象在InputFormat后面。Flink提供了几种内置格式，可以从常见的文件格式创建数据集。它们中的许多在ExecutionEnvironment上都有快捷方法。

1、基于文件

readTextFile(path) / TextInputFormat ：读取文本文件。 readTextFileWithValue(path) / TextValueInputFormat： 读取文件，并将它们作为StringValues返回。StringValues是可变字符串。 readCsvFile(path) / CsvInputFormat：解析带有逗号(或其他字符)分隔字段的文件。返回由元组或pojo组成的数据集。支持基本java类型及其对应值作为字段类型。 readFileOfPrimitives(path, Class) / PrimitiveInputFormat：解析以新行(或另一个字符序列)分隔的原始数据类型(如String或Integer)的文件。 readFileOfPrimitives(path, delimiter, Class) / PrimitiveInputFormat：使用给定的分隔符解析以新行(或另一个字符序列)分隔的原始数据类型(如String或Integer)的文件。

2、基于集合

fromCollection(Collection)：从Java.util.Collection创建一个数据集。集合中的所有元素必须具有相同的类型。 fromCollection(Iterator, Class)：从迭代器创建数据集。该类指定迭代器返回的元素的数据类型。 fromElements(T …)：根据给定的对象序列创建一个数据集。所有对象必须是相同的类型。 fromParallelCollection(SplittableIterator, Class)：并行地从迭代器创建数据集。该类指定迭代器返回的元素的数据类型。 generateSequence(from, to)：并行生成给定区间内的数字序列。 3、通用型 readFile(inputFormat, path) / FileInputFormat ：接受文件输入格式。 createInput(inputFormat) / InputFormat：接受通用输入格式。

2）DataSet Transformations（数据集转换//处理/算子）

数据转换将一个或多个数据集转换为新的数据集。程序可以将多个转换组合成复杂的程序集。

3）Data Sinks（数据输出）

数据接收器使用数据集，并用于存储或返回它们。使用OutputFormat描述数据接收器操作。Flink提供了多种内置的输出格式，这些格式封装在DataSet上的操作后面：

writeAsText() / TextOutputFormat：按行方式将元素写入字符串。字符串是通过调用每个元素的toString()方法获得的。 writeAsFormattedText() / TextOutputFormat：将元素按行编写为字符串。字符串是通过为每个元素调用用户定义的format()方法获得的。 writeAsCsv(…) / CsvOutputFormat：将元组写入逗号分隔的值文件。行和字段分隔符是可配置的。每个字段的值来自对象的toString()方法。print() / printToErr() / print(String msg) / printToErr(String msg) -打印出标准输出/标准错误流中每个元素的toString()值。可选地，可以提供一个前缀(msg)，作为输出的前缀。这有助于区分不同的打印调用。如果并行度大于1，输出也会被添加产生输出的任务的标识符。 write() / FileOutputFormat：方法和基类用于自定义文件输出。支持自定义对象到字节的转换。 output()/ OutputFormat：大多数通用输出方法，用于非基于文件的数据接收器(例如将结果存储在数据库中)。

一个数据集可以被输入到多个操作。程序可以写或打印一个数据集，同时在它们上运行额外的转换。

【示例】

package com import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment} import org.apache.flink.core.fs.FileSystem.WriteMode object DataSetTest001 { def main(args: Array[String]): Unit = { val env = ExecutionEnvironment.getExecutionEnvironment env.setParallelism(1) // text data val textData: DataSet[String] = env.readTextFile("flink/data/s1") // write DataSet to a file on the local file system // textData.writeAsText("flink/data/sink01") // write DataSet to a file on an HDFS with a namenode running at nnHost:nnPort // 先创建目录：hadoop fs -mkdir -p hdfs://hadoop-node1:8082/flink/DataSet/ // 操作添加依赖 /* org.apache.hadoop hadoop-hdfs 3.3.1 provided */ textData.writeAsText("hdfs://hadoop-node1:8082/flink/DataSet/sink02") // // // write DataSet to a file and overwrite the file if it exists // textData.writeAsText("flink/data/sink03", WriteMode.OVERWRITE) // // // tuples as lines with pipe as the separator "a|b|c" // val values: DataSet[(String, Int, Double)] = // [...] // values.writeAsCsv("file:///path/to/the/result/file", "\n", "|") // // // this writes tuples in the text formatting "(a, b, c)", rather than as CSV lines // values.writeAsText("file:///path/to/the/result/file") // this writes values as strings using a user-defined formatting // values map { tuple => tuple._1 + " - " + tuple._2 } // .writeAsText("file:///path/to/the/result/file") env.execute("dataset test") } }

【示例】WordCount

package com import org.apache.flink.api.scala._ object WordCount { def main(args: Array[String]): Unit = { val env = ExecutionEnvironment.getExecutionEnvironment val text = env.fromElements( "Who's there?", "I think I hear them. Stand, ho! Who's there?") val counts = text .flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } } .map { (_, 1) } .groupBy(0) .sum(1) counts.print() } }

未完待续，更多大数据知识，请耐心等待~

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。