• 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 返回。

    • 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

    • 可能用得到的依赖

    1.   <dependencies>
    2.     
    3.     <dependency>
    4.       <groupId>org.projectlombokgroupId>
    5.       <artifactId>lombokartifactId>
    6.       <version>1.18.20version>
    7.       <scope>providedscope>
    8.     dependency>
    9.  
    10.     <dependency>
    11.       <groupId>org.scala-langgroupId>
    12.       <artifactId>scala-libraryartifactId>
    13.       <version>${scala.version}version>
    14.     dependency>
    15.     
    16.     <dependency>
    17.       <groupId>org.scala-langgroupId>
    18.       <artifactId>scala-reflectartifactId>
    19.       <version>${scala.version}version>
    20.     dependency>
    21.     <dependency>
    22.       <groupId>org.scala-langgroupId>
    23.       <artifactId>scala-compilerartifactId>
    24.       <version>${scala.version}version>
    25.     dependency>
    26.  
    27.     <dependency>
    28.       <groupId>org.apache.flinkgroupId>
    29.       <artifactId>flink-scala_${scala.binary.version}artifactId>
    30.       <version>${flink.version}version>
    31.     dependency>
    32.     <dependency>
    33.       <groupId>org.apache.flinkgroupId>
    34.       <artifactId>flink-streaming-scala_${scala.binary.version}artifactId>
    35.       <version>${flink.version}version>
    36.     dependency>
    37.  
    38.     <dependency>
    39.       <groupId>org.apache.flinkgroupId>
    40.       <artifactId>flink-clients_${scala.binary.version}artifactId>
    41.       <version>${flink.version}version>
    42.     dependency>
    43.     <dependency>
    44.       <groupId>org.apache.kafkagroupId>
    45.       <artifactId>kafka-clientsartifactId>
    46.       <version>${kafka.version}version>
    47.     dependency>
    48.     <dependency>
    49.       <groupId>org.apache.flinkgroupId>
    50.       <artifactId>flink-connector-kafka_${scala.binary.version}artifactId>
    51.       <version>${flink.version}version>
    52.     dependency>
    53.     <dependency>
    54.       <groupId>org.apache.hadoopgroupId>
    55.       <artifactId>hadoop-clientartifactId>
    56.       <version>${hadoop.version}version>
    57.     dependency>
    58.     <dependency>
    59.       <groupId>mysqlgroupId>
    60.       <artifactId>mysql-connector-javaartifactId>
    61.       <version>${mysql.version}version>
    62.     dependency>
    63.     <dependency>
    64.       <groupId>com.alibabagroupId>
    65.       <artifactId>fastjsonartifactId>
    66.       <version>${fastjson.version}version>
    67.     dependency>
    68.  
    69.     
    70.     <dependency>
    71.       <groupId>org.apache.flinkgroupId>
    72.       <artifactId>flink-table-api-scala-bridge_${scala.binary.version}artifactId>
    73.       <version>${flink.version}version>
    74.       <scope>providedscope>
    75.     dependency>
    76.     <dependency>
    77.       <groupId>org.apache.flinkgroupId>
    78.       <artifactId>flink-table-planner-blink_${scala.binary.version}artifactId>
    79.       <version>${flink.version}version>
    80.       <scope>providedscope>
    81.     dependency>
    82.     <dependency>
    83.       <groupId>org.apache.flinkgroupId>
    84.       <artifactId>flink-table-commonartifactId>
    85.       <version>${flink.version}version>
    86.       <scope>providedscope>
    87.     dependency>
    88.  
    89.     <dependency>
    90.       <groupId>org.apache.flinkgroupId>
    91.       <artifactId>flink-csvartifactId>
    92.       <version>${flink.version}version>
    93.     dependency>
    94.  
    95.     <dependency>
    96.       <groupId>org.apache.flinkgroupId>
    97.       <artifactId>flink-connector-baseartifactId>
    98.       <version>${flink.version}version>
    99.     dependency>
    100.  
    101.     <dependency>
    102.       <groupId>org.apache.flinkgroupId>
    103.       <artifactId>flink-javaartifactId>
    104.       <version>1.13.2version>
    105.     dependency>
    106.     <dependency>
    107.       <groupId>org.apache.flinkgroupId>
    108.       <artifactId>flink-streaming-java_2.12artifactId>
    109.       <version>1.13.2version>
    110.     dependency>
    111.     <dependency>
    112.       <groupId>org.apache.flinkgroupId>
    113.       <artifactId>flink-table-api-java-bridge_2.12artifactId>
    114.       <version>1.13.2version>
    115.     dependency>
    116.  
    117.     <dependency>
    118.       <groupId>com.typesafe.playgroupId>
    119.       <artifactId>play-json_2.12artifactId>
    120.       <version>2.9.2version>
    121.     dependency>
    122.   dependencies>

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


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

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

    具体 代码如下
     

    1. package stream
    2.  
    3. import org.apache.flink.api.common.functions.ReduceFunction
    4. import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
    5. import org.apache.flink.streaming.api.functions.source.SourceFunction
    6. import org.apache.flink.streaming.api.scala._
    7. import org.apache.flink.streaming.api.watermark.Watermark
    8. import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
    9. import org.apache.flink.streaming.api.windowing.time.Time
    10. import scala.util.Random
    11.  
    12. object FlinkRandomTemperatureSource {
    13.   private case class TemperatureReading(timestamp: Long, sensorId: String, temperature: Double)
    14.  
    15.   private class RandomTemperatureSource(var totalElements: Int) extends SourceFunction[TemperatureReading] {
    16.     private var isRunning = true
    17.     private val random = new Random()
    18.     private var currentTime = System.currentTimeMillis()
    19.     private var baseTemperature = 30.0
    20.  
    21.     override def run(ctx: SourceFunction.SourceContext[TemperatureReading]): Unit = {
    22.       while (isRunning && totalElements > 0) {
    23.         val delay = (random.nextDouble() * 1000 + 500).toLong // 0.5到1.5秒之间的随机延迟
    24.         Thread.sleep(delay)
    25.  
    26.         // 模拟时间流逝和温度增加
    27.         currentTime += delay
    28.         baseTemperature += delay / 1000000.0
    29.  
    30.         // 生成并发送数据
    31.         val sensorId = "sensor" + random.nextInt(10)
    32.         val temperatureReading = TemperatureReading(currentTime, sensorId, baseTemperature)
    33.         ctx.collect(temperatureReading)
    34.  
    35.         totalElements -= 1
    36.       }
    37.  
    38.       ctx.close()
    39.     }
    40.  
    41.     override def cancel(): Unit = {
    42.       isRunning = false
    43.     }
    44.   }
    45.  
    46.   /**
    47.    * 自定义时间戳和水位线分配器
    48.    * AssignerWithPunctuatedWatermarks其实已经过时了
    49.    * 建议使用assignTimestampsAndWatermarks
    50.    */
    51.   private class TemperatureTimestampExtractor extends AssignerWithPunctuatedWatermarks[TemperatureReading] {
    52.     override def extractTimestamp(element: TemperatureReading, previousElementTimestamp: Long): Long = {
    53.       element.timestamp
    54.     }
    55.  
    56.     override def checkAndGetNextWatermark(lastElement: TemperatureReading, extractedTimestamp: Long): Watermark = {
    57.       new Watermark(extractedTimestamp - 1) // 简单地，我们可以将水位线设置为当前时间戳减1
    58.     }
    59.   }
    60.  
    61.   def main(args: Array[String]): Unit = {
    62.     val env = StreamExecutionEnvironment.getExecutionEnvironment
    63.  
    64.  
    65.     val source = new RandomTemperatureSource(100000) // 生成100000条数据
    66.     val stream = env.addSource(source)
    67.  
    68.     // 为数据源分配时间戳和水位线
    69.     val timestampedStream = stream.assignTimestampsAndWatermarks(new TemperatureTimestampExtractor)
    70.  
    71.     timestampedStream
    72.       .keyBy(_.sensorId)
    73.       .window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
    74.       .reduce((val1,val2)=>TemperatureReading
    75.       (val1.timestamp,val1.sensorId,(val1.temperature+val2.temperature)/2))
    76. //      .reduce(new ReduceFunction[TemperatureReading] {
    77. //        override def reduce(value1: TemperatureReading,
    78.       //        value2: TemperatureReading): TemperatureReading = {
    79. //          val avgTemp = (value1.temperature + value2.temperature) /2
    80. //          TemperatureReading(value1.timestamp, value1.sensorId, avgTemp)
    81. //        }
    82. //      })
    83.       .print()
    84.     env.execute("Flink Random Temperature Source with Event Time")
    85.  
    86.   }
    87. }

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

  • 相关阅读:
    mysql笔记
    区间重叠问题
    CVE-2021-3560 Linux Polkit 权限提升漏洞
    水果店圈子：现在开水果店需要什么设备，开水果店所需要的设备有哪些
    centos7系统下，实现1台服务器免密登录多台服务器功能
    NodeRed系列——websoket展示数据
    工业控制系统与传统信息系统安全的对比分析
    Django--19开发用例系列功能
    ASP.NET Core 6框架揭秘实例演示[17]：利用IHttpClientFactory工厂来创建HttpClient
    华为发布应用流程
  • 原文地址：https://blog.csdn.net/zzy66666c/article/details/141056058