大数据学习之Flink算子、了解(Source)源算子(基础篇二)

发布时间:2024年01月23日

Source源算子(基础篇二)


目录

Source源算子(基础篇二)

二、源算子(source)

1. 准备工作

2.从集合中读取数据

可以使用代码中的fromCollection()方法直接读取列表

也可以使用代码中的fromElements()方法直接列出数据获取

3. 从文件中读取数据

说明:

4. 从Socket读取数据

(1)编写StreamWordCount

(2)在 Linux 环境的主机bigdata1 上,执行下列命令,发送数据进行测试:

(3)启动 StreamWordCount 程序

(4)从 bigdata1 发送数据:

(5)看控制台的输出结果

5.从Kafka读取数据

6.自定义源算子(source)

7.Flink支持的数据类型


二、源算子(source)

Flink 可以从各种来源获取数据,然后构建 DataStream 进行转换处理。一般将数据的输入 来源称为数据源,而读取数据的算子就是源算子(Source)。所以,Source 就是我们整个处理 程序的输入端。

Flink 代码中通用的添加 Source 的方式,是调用执行环境的 addSource()方法:

//通过调用 addSource()方法可以获取 DataStream 对象
val stream = env.addSource(...)

方法传入一个对象参数,需要实现 SourceFunction 接口,返回一个 DataStream。

1. 准备工作

case class Event(user: String, url: String, timestamp: Long)

2.从集合中读取数据

  • 最简单的读取数据的方式,就是在代码中直接创建一个集合,然后调用执行环境的 fromCollection 方法进行读取。
  • 这相当于将数据临时存储到内存中,形成特殊的数据结构后, 作为数据源使用,一般用于测试。
import org.apache.flink.streaming.api.scala._

case class Event(user: String, url: String, timestamp: Long)

object SourceCollection {
  def main(args: Array[String]): Unit = {
    //获取流执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //设置并行度(并行任务的数量)为1
    env.setParallelism(1)
    // 创建包含点击事件的列表
    // 点击操作中包含两个事件
    val clicks = List(Event("Mary", "/.home", 1000L), Event("Bob", "/.cart", 2000L))
    //将列表作为流输出
    //把clicks作为数据流
    val stream = env.fromCollection(clicks)
    //fromElements从给定的元素集合中创建一个DataStream
    val stream1 = env.fromElements(
      Event("zhangsan","/.opt",1000L),
      Event("lisi","/.opt",2000L)
    )
    stream.print("stream")
    stream1.print("stream1")
    env.execute()
  }
}

可以使用代码中的fromCollection()方法直接读取列表

也可以使用代码中的fromElements()方法直接列出数据获取

3. 从文件中读取数据

真正的实际应用中,自然不会直接将数据写在代码中。通常情况下,我们会从存储介质中 获取数据,一个比较常见的方式就是读取日志文件。这也是批处理中最常见的读取方式。

val stream = env.readTextFile("input/words.txt")
说明:
  • 参数可以是文件,可以是目录

  • 可以是绝对路径,也可以是相对路径

  • 相对路径是从系统属性 user.dir 获取路径:在 IDEA 下是 project 的根目录, standalone 模式下是集群节点根目录;

    • 系统属性 user.dir:这是一个Java系统属性,它表示用户当前的工作目录。在很多应用中,它通常被用作参考路径。

    • IDEA下是project的根目录:当你在IDEA中打开一个项目时,项目的根目录通常是IDEA的工作目录。相对路径就是基于这个根目录来确定的。

    • standalone模式下是集群节点根目录:如Hadoop分布式计算系统中的独立模式(standalone mode)。在这种模式下,路径可能是相对于集群节点的根目录。

  • 也可以从 HDFS 目录下读取, 使用路径 hdfs://...

    • 前提要在pom文件中添加hadoop相关依赖

4. 从Socket读取数据

不论从集合还是文件,我们读取的其实都是有界数据。在流处理的场景中,数据往往是无 界的。一个简单的例子,就是我们之前用到的读取 socket 文本流。这种方式由于吞吐量小、 稳定性较差,一般也是用于测试。

//通过主机名和端口号读取socket文本流
    val linDs = env.socketTextStream("bigdata1",7777)

具体实现案例:

(1)编写StreamWordCount

import org.apache.flink.streaming.api.scala._

object StreamWordCount {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //通过主机名和端口号读取socket文本流
    val linDs = env.socketTextStream("bigdata1",7777)
    //进行转换计算
    val result = linDs
      .flatMap(data => data.split(" ")) //用空格切分字符串
      .map((_,1)) //切分后的字符串转换为一个元组
      .keyBy(_._1) //使用元组的第一个字段进行分组
      .sum(1) //分组后的数据的第二个字段进行累加
    //打印计算结果
    result.print()
    env.execute()
  }
}

(2)在 Linux 环境的主机bigdata1 上,执行下列命令,发送数据进行测试:

$ nc -lk 7777

(3)启动 StreamWordCount 程序

我们会发现程序启动之后没有任何输出、也不会退出。这是正常的——因为 Flink 的流处 理是事件驱动的,当前程序会一直处于监听状态,只有接收到数据才会执行任务、输出统计结果。

(4)从 bigdata1 发送数据:

hello flink
hello world
hello scala

(5)看控制台的输出结果

5.从Kafka读取数据

Kafka 作为分布式消息传输队列,是一个高吞吐、易于扩展的消息系统。

而消息队列的传输方式,恰恰和流处理是完全一致的。

所以可以说 Kafka 和 Flink 天生一对,是当前处理流式数据的双子星。

在如今的实时流处理应用中,由 Kafka 进行数据的收集和传输,Flink 进行分析计算,这样的架构已经成为众多企业的首选

调用 env.addSource(),传入 FlinkKafkaConsumer 的对象实例就可以了。

创建 FlinkKafkaConsumer 时需要传入三个参数:

  • 第一个参数 topic,定义了从哪些主题中读取数据。可以是一个 topic,也可以是 topic 列表,还可以是匹配所有想要读取的 topic 的正则表达式。当从多个 topic 中读取数据 时,Kafka 连接器将会处理所有 topic 的分区,将这些分区的数据放到一条数据流中 去。
  • 第二个参数是一个 DeserializationSchema 或者 KeyedDeserializationSchema。Kafka 消 息被存储为原始的字节数据,所以需要反序列化成 Java 或者 Scala 对象。上面代码中 53 使用的 SimpleStringSchema,是一个内置的 DeserializationSchema,它只是将字节数 组简单地反序列化成字符串。DeserializationSchema 和 KeyedDeserializationSchema 是 公共接口,所以我们也可以自定义反序列化逻辑。
  • 第三个参数是一个 Properties 对象,设置了 Kafka 客户端的一些属性。
更新中...

6.自定义源算子(source)

接下来我们创建一个自定义的数据源,实现 SourceFunction 接口。主要重写两个关键方法: run()和 cancel()。

  • run()方法:使用运行时上下文对象(SourceContext)向下游发送数据;
  • cancel()方法:通过标识位控制退出循环,来达到中断数据源的效果。

7.Flink支持的数据类型

文章来源:https://blog.csdn.net/2301_78038072/article/details/135762741
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。