0
0

Flink入门 (二)--Flink程序的编写

其他案例demo可以参考我的GitHub

https://github.com/NuistGeorgeYoung/flink_stream_test/

编写一个Flink程序大致上可以分为以下几个步骤：

  • 获得一个execution environment， env

val env = StreamExecutionEnvironment.getExecutionEnvironment

之后你可以设置以下配置

  1. 并行度（Parallelism）：- 设置作业的并行度，即操作符（operator）的并行实例数。这可以通过env.setParallelism(int parallelism)来设置。

  2. 执行模式（Execution Mode）：- Flink支持两种执行模式：本地执行（LOCAL）和集群执行（CLUSTER）。虽然getExecutionEnvironment默认可能根据你的环境选择执行模式，但在某些情况下你可能需要显式设置。不过，通常这不是通过StreamExecutionEnvironment直接设置的，而是通过配置文件或命令行参数来控制的。

  3. 时间特性（Time Characteristics）：- Flink支持处理时间（Processing Time）、事件时间（Event Time）和摄入时间（Ingestion Time）。你可以通过env.setStreamTimeCharacteristic(TimeCharacteristic timeCharacteristic)来设置时间特性。

  4. 状态后端（State Backend）：- 状态后端用于存储和管理Flink作业的状态。你可以通过env.setStateBackend(StateBackend backend)来设置状态后端。常见的状态后端包括FsStateBackend（基于文件系统的状态后端）和RocksDBStateBackend（基于RocksDB的状态后端）。

  5. 检查点（Checkpointing）：- Flink的容错机制依赖于检查点（Checkpointing）。你可以通过env.enableCheckpointing(long interval)来启用检查点，并通过一系列其他方法来配置检查点的具体行为，如setCheckpointingMode、setCheckpointTimeout等。

  6. 重启策略（Restart Strategies）：- Flink允许你配置作业在遇到故障时的重启策略。你可以通过env.setRestartStrategy方法来设置重启策略，比如固定延迟重启、失败率重启等。

  7. 任务槽（Task Slots）和资源（Resources）：- 这些配置通常不是通过StreamExecutionEnvironment直接设置的，而是通过Flink的配置文件（如flink-conf.yaml）或提交作业时的命令行参数来配置的。它们涉及到Flink集群的资源分配和任务调度。

基于集合

通用方法

  • 加载/创建初始数据， source

  • 数据集包括但不限于以下这几种

  • 基于文件

  • readTextFile(path)/ TextInputFormat- 按行读取文件并将其作为字符串返回。
  • readTextFileWithValue(path)/ TextValueInputFormat- 按行读取文件并将它们作为StringValues返回。StringValues是可变字符串。
  • readCsvFile(path)/ CsvInputFormat- 解析逗号（或其他字符）分隔字段的文件。返回元组或POJO的DataSet。支持基本java类型及其Value对应作为字段类型。
  • readFileOfPrimitives(path, Class)/ PrimitiveInputFormat- 解析新行（或其他字符序列）分隔的原始数据类型（如String或）的文件Integer。
  • readFileOfPrimitives(path, delimiter, Class)/ PrimitiveInputFormat- 解析新行（或其他字符序列）分隔的原始数据类型的文件，例如String或Integer使用给定的分隔符。
  • readSequenceFile(Key, Value, path)/ SequenceFileInputFormat- 创建一个JobConf并从类型为SequenceFileInputFormat，Key class和Value类的指定路径中读取文件，并将它们作为Tuple2 <Key，Value>返回。
  • fromCollection(Collection) - 从Java Java.util.Collection创建数据集。集合中的所有数据元必须属于同一类型。
  • fromCollection(Iterator, Class) - 从迭代器创建数据集。该类指定迭代器返回的数据元的数据类型。
  • fromElements(T ...) - 根据给定的对象序列创建数据集。所有对象必须属于同一类型。
  • fromParallelCollection(SplittableIterator, Class)- 并行地从迭代器创建数据集。该类指定迭代器返回的数据元的数据类型。
  • generateSequence(from, to) - 并行生成给定间隔中的数字序列。
  • readFile(inputFormat, path)/ FileInputFormat- 接受文件输入格式。
  • createInput(inputFormat)/ InputFormat- 接受通用输入格式。

  • 指定此数据的转换， transform

  • 指定放置计算结果的位置（例如输出）， sink

    方法名对应的OutputFormat类描述

    writeAsText()


TextOutputFormat

按字符串顺序写入数据元。通过调用每个数据元的

toString()

方法获得字符串。

writeAsFormattedText()


TextOutputFormat

按字符串顺序写数据元。通过为每个数据元调用用户定义的

format()

方法来获取字符串。

writeAsCsv(...)


CsvOutputFormat

将元组写为逗号分隔值文件。行和字段分隔符是可配置的。每个字段的值来自对象的

toString()

方法。

print()

-

在标准输出上打印每个数据元的

toString()

值。

printToErr()

-

在标准错误流上打印每个数据元的

toString()

值。

print(String msg)

-

在标准输出上打印每个数据元的

toString()

值，并可选地提供一个前缀（msg）。

printToErr(String msg)

-

在标准错误流上打印每个数据元的

toString()

值，并可选地提供一个前缀（msg）。

write()


FileOutputFormat

自定义文件输出的方法和基类。支持自定义对象到字节的转换。

output()


OutputFormat

（泛型）
大多数通用输出方法，用于非基于文件的数据接收器（例如将结果存储在数据库中）。需要具体实现

OutputFormat

接口。
注意：

print()

、

printToErr()

、

print(String msg)

和 

printToErr(String msg)

方法并不直接关联到某个特定的

OutputFormat

类，因为它们是用于调试目的的直接输出到控制台的方法。同样，

output()

方法是一个更通用的接口，它依赖于具体的

OutputFormat

实现来定义数据如何被输出。

  • 触发程序执行 ， execute

  • 实战DEMO

  • 可能用得到的依赖

  <dependencies>
    <!-- lombok -->
    <dependency>
      <groupId>org.projectlombok</groupId>
      <artifactId>lombok</artifactId>
      <version>1.18.20</version>
      <scope>provided</scope>
    </dependency>

    <dependency>
      <groupId>org.scala-lang</groupId>
      <artifactId>scala-library</artifactId>
      <version>${scala.version}</version>
    </dependency>
    <!--  scala reflect  -->
    <dependency>
      <groupId>org.scala-lang</groupId>
      <artifactId>scala-reflect</artifactId>
      <version>${scala.version}</version>
    </dependency>
    <dependency>
      <groupId>org.scala-lang</groupId>
      <artifactId>scala-compiler</artifactId>
      <version>${scala.version}</version>
    </dependency>

    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-scala_${scala.binary.version}</artifactId>
      <version>${flink.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
      <version>${flink.version}</version>
    </dependency>

    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-clients_${scala.binary.version}</artifactId>
      <version>${flink.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka-clients</artifactId>
      <version>${kafka.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
      <version>${flink.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-client</artifactId>
      <version>${hadoop.version}</version>
    </dependency>
    <dependency>
      <groupId>mysql</groupId>
      <artifactId>mysql-connector-java</artifactId>
      <version>${mysql.version}</version>
    </dependency>
    <dependency>
      <groupId>com.alibaba</groupId>
      <artifactId>fastjson</artifactId>
      <version>${fastjson.version}</version>
    </dependency>

    <!--  flink table  -->
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-api-scala-bridge_${scala.binary.version}</artifactId>
      <version>${flink.version}</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
      <version>${flink.version}</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-common</artifactId>
      <version>${flink.version}</version>
      <scope>provided</scope>
    </dependency>

    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-csv</artifactId>
      <version>${flink.version}</version>
    </dependency>

    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-base</artifactId>
      <version>${flink.version}</version>
    </dependency>

    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-java</artifactId>
      <version>1.13.2</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-streaming-java_2.12</artifactId>
      <version>1.13.2</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-table-api-java-bridge_2.12</artifactId>
      <version>1.13.2</version>
    </dependency>

    <dependency>
      <groupId>com.typesafe.play</groupId>
      <artifactId>play-json_2.12</artifactId>
      <version>2.9.2</version>
    </dependency>
  </dependencies>

我通过run方法不断模拟生成温度读数（TemperatureReading），每个读数包含时间戳、传感器ID和温度值。
通过随机延迟来模拟数据到达的时间间隔，同时温度值也随时间（以毫秒为单位）的增加而缓慢上升。

**

TemperatureTimestampExtractor

**负责为每个温度读数分配时间戳，并生成水位线。在这个例子中，水位线简单地设置为当前时间戳减1，这在实际应用中可能不是最佳实践，仅作为学习交流时的粗略设定。
定义了一个基于处理时间的滚动窗口，窗口大小为1秒。
在窗口内，使用reduce方法计算温度的平均值。

最后对结果进行简单的println()输出。

具体 代码如下

package stream

import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import scala.util.Random

object FlinkRandomTemperatureSource {
  private case class TemperatureReading(timestamp: Long, sensorId: String, temperature: Double)

  private class RandomTemperatureSource(var totalElements: Int) extends SourceFunction[TemperatureReading] {
    private var isRunning = true
    private val random = new Random()
    private var currentTime = System.currentTimeMillis()
    private var baseTemperature = 30.0

    override def run(ctx: SourceFunction.SourceContext[TemperatureReading]): Unit = {
      while (isRunning && totalElements > 0) {
        val delay = (random.nextDouble() * 1000 + 500).toLong // 0.5到1.5秒之间的随机延迟
        Thread.sleep(delay)

        // 模拟时间流逝和温度增加
        currentTime += delay
        baseTemperature += delay / 1000000.0

        // 生成并发送数据
        val sensorId = "sensor" + random.nextInt(10)
        val temperatureReading = TemperatureReading(currentTime, sensorId, baseTemperature)
        ctx.collect(temperatureReading)

        totalElements -= 1
      }

      ctx.close()
    }

    override def cancel(): Unit = {
      isRunning = false
    }
  }

  /**
   * 自定义时间戳和水位线分配器
   * AssignerWithPunctuatedWatermarks其实已经过时了
   * 建议使用assignTimestampsAndWatermarks
   */
  private class TemperatureTimestampExtractor extends AssignerWithPunctuatedWatermarks[TemperatureReading] {
    override def extractTimestamp(element: TemperatureReading, previousElementTimestamp: Long): Long = {
      element.timestamp
    }

    override def checkAndGetNextWatermark(lastElement: TemperatureReading, extractedTimestamp: Long): Watermark = {
      new Watermark(extractedTimestamp - 1) // 简单地，我们可以将水位线设置为当前时间戳减1
    }
  }

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val source = new RandomTemperatureSource(100000) // 生成100000条数据
    val stream = env.addSource(source)

    // 为数据源分配时间戳和水位线
    val timestampedStream = stream.assignTimestampsAndWatermarks(new TemperatureTimestampExtractor)

    timestampedStream
      .keyBy(_.sensorId)
      .window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
      .reduce((val1,val2)=>TemperatureReading
      (val1.timestamp,val1.sensorId,(val1.temperature+val2.temperature)/2))
//      .reduce(new ReduceFunction[TemperatureReading] {
//        override def reduce(value1: TemperatureReading,
      //        value2: TemperatureReading): TemperatureReading = {
//          val avgTemp = (value1.temperature + value2.temperature) /2
//          TemperatureReading(value1.timestamp, value1.sensorId, avgTemp)
//        }
//      })
      .print()
    env.execute("Flink Random Temperature Source with Event Time")

  }
}

最后运行程序，会持续输出结果，模拟夏天上午的温度缓慢上升的趋势

标签： flink 大数据 scala
本文转载自: https://blog.csdn.net/zzy66666c/article/details/141056058
版权归原作者 不秀不亏不怼不皮 所有， 如有侵权，请联系我们删除。
登录后发布评论

“Flink入门 (二)--Flink程序的编写”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

高级网络安全——IP 安全（week5）

KeyloggerV1.0：安全的键盘与鼠标记录工具

RabbitMQ如何避免消息重复投递或重复消费？

【Vue3】 中的 【unref】：详解与使用

[前端]NodeJS常见面试题目

第五章springboot实现web的常用功能

SpringMVC详解

文章导航