Flink_DataStreamAPI_源算子Source

ops/2024/11/17 8:11:00/

Flink_DataStreamAPI_源算子Source

  • 1从集合中读取数据
  • 2从文件读取数据
  • 3从Socket读取数据
  • 4从Kafka读取数据
  • 5从数据生成器读取数据
  • Flink支持的数据类型
    • 1)Flink的类型系统
    • 2)Flink支持的数据类型
    • 3)类型提示(Type Hints)

1从集合中读取数据

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

public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();List<Integer> data = Arrays.asList(1, 22, 3);DataStreamSource<Integer> ds = env.fromCollection(data);stream.print();env.execute();
}

2从文件读取数据

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

 <dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-files</artifactId><version>${flink.version}</version>
</dependency>
public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();FileSource<String> fileSource = FileSource.forRecordStreamFormat(new TextLineInputFormat(), new Path("input/word.txt")).build();env.fromSource(fileSource,WatermarkStrategy.noWatermarks(),"file").print();env.execute();
}

说明:
参数可以是目录,也可以是文件;还可以从HDFS目录下读取,使用路径hdfs://…;
路径可以是相对路径,也可以是绝对路径;
相对路径是从系统属性user.dir获取路径:idea下是project的根目录,standalone模式下是集群节点根目录;

3从Socket读取数据

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

DataStream<String> stream = env.socketTextStream("localhost", 7777);

4从Kafka读取数据

Flink官方提供了连接工具flink-connector-kafka,直接帮我们实现了一个消费者FlinkKafkaConsumer,它就是用来读取Kafka数据的SourceFunction。
所以想要以Kafka作为数据源获取数据,我们只需要引入Kafka连接器的依赖。Flink官方提供的是一个通用的Kafka连接器,它会自动跟踪最新版本的Kafka客户端。目前最新版本只支持0.10.0版本以上的Kafka。这里我们需要导入的依赖如下。

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka</artifactId><version>${flink.version}</version>
</dependency>
public class SourceKafka {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();KafkaSource<String> kafkaSource = KafkaSource.<String>builder().setBootstrapServers("hadoop102:9092").setTopics("topic_1").setGroupId("atguigu").setStartingOffsets(OffsetsInitializer.latest()).setValueOnlyDeserializer(new SimpleStringSchema()) .build();DataStreamSource<String> stream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka-source");stream.print("Kafka");env.execute();}
}

5从数据生成器读取数据

Flink从1.11开始提供了一个内置的DataGen 连接器,主要是用于生成一些随机数,用于在没有数据源的时候,进行流任务的测试以及性能测试等。1.17提供了新的Source写法,需要导入依赖:

        <dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-datagen</artifactId><version>${flink.version}</version></dependency>
public class DataGeneratorDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataGeneratorSource<String> dataGeneratorSource =new DataGeneratorSource<>(new GeneratorFunction<Long, String>() {@Overridepublic String map(Long value) throws Exception {return "Number:"+value;}},Long.MAX_VALUE,RateLimiterStrategy.perSecond(10),Types.STRING);env.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(), "datagenerator").print();env.execute();}
}

Flink支持的数据类型

1)Flink的类型系统

Flink使用“类型信息”(TypeInformation)来统一表示数据类型。TypeInformation类是Flink中所有类型描述符的基类。它涵盖了类型的一些基本属性,并为每个数据类型生成特定的序列化器、反序列化器和比较器。

2)Flink支持的数据类型

对于常见的Java和Scala数据类型,Flink都是支持的。Flink在内部,Flink对支持不同的类型进行了划分,这些类型可以在Types工具类中找到:

(1)基本类型
所有Java基本类型及其包装类,再加上Void、String、Date、BigDecimal和BigInteger。

(2)数组类型
包括基本类型数组(PRIMITIVE_ARRAY)和对象数组(OBJECT_ARRAY)。

(3)复合数据类型
-Java元组类型(TUPLE):这是Flink内置的元组类型,是Java API的一部分。最多25个字段,也就是从Tuple0~Tuple25,不支持空字段。
-Scala 样例类及Scala元组:不支持空字段。
-行类型(ROW):可以认为是具有任意个字段的元组,并支持空字段。
-POJO:Flink自定义的类似于Java bean模式的类。

(4)辅助类型
Option、Either、List、Map等。

(5)泛型类型(GENERIC)
Flink支持所有的Java类和Scala类。不过如果没有按照上面POJO类型的要求来定义,就会被Flink当作泛型类来处理。Flink会把泛型类型当作黑盒,无法获取它们内部的属性;它们也不是由Flink本身序列化的,而是由Kryo序列化的。
在这些类型中,元组类型和POJO类型最为灵活,因为它们支持创建复杂类型。而相比之下,POJO还支持在键(key)的定义中直接使用字段名,这会让我们的代码可读性大大增加。所以,在项目实践中,往往会将流处理程序中的元素类型定为Flink的POJO类型。
Flink对POJO类型的要求如下:
-类是公有(public)的
-有一个无参的构造方法
-所有属性都是公有(public)的
-所有属性的类型都是可以序列化的

3)类型提示(Type Hints)

Flink还具有一个类型提取系统,可以分析函数的输入和返回类型,自动获取类型信息,从而获得对应的序列化器和反序列化器。但是,由于Java中泛型擦除的存在,在某些特殊情况下(比如Lambda表达式中),自动提取的信息是不够精细的——只告诉Flink当前的元素由“船头、船身、船尾”构成,根本无法重建出“大船”的模样;这时就需要显式地提供类型信息,才能使应用程序正常工作或提高其性能。
为了解决这类问题,Java API提供了专门的“类型提示”(type hints)。
回忆一下之前的word count流处理程序,我们在将String类型的每个词转换成(word, count)二元组后,就明确地用returns指定了返回的类型。因为对于map里传入的Lambda表达式,系统只能推断出返回的是Tuple2类型,而无法得到Tuple2<String, Long>。只有显式地告诉系统当前的返回类型,才能正确地解析出完整数据。

.map(word -> Tuple2.of(word, 1L))
.returns(Types.TUPLE(Types.STRING, Types.LONG));

Flink还专门提供了TypeHint类,它可以捕获泛型的类型信息,并且一直记录下来,为运行时提供足够的信息。我们同样可以通过.returns()方法,明确地指定转换之后的DataStream里元素的类型。

returns(new TypeHint<Tuple2<Integer, SomeType>>(){})

http://www.ppmy.cn/ops/134370.html

相关文章

springboot整合security5.7.16实现用户登录及超时自动登录

1、问题概述? 提供源码下载,springSecurity关键技术,仔细阅读。 解决如下问题: 1、使用springboot2.7.16整合springSecurity5.7.16实现用户登录 2、解决新版本的Security的配置文件-提供完整配置。 3、当用户登录信息失效后,在界面发送的ajax,如何重定向到login.htm…

Android Framework与JNI

本文以 android-12.0.0_r34 的代码进行分析。 framework中的JNI 通常来说&#xff0c;Android framework 中使用到的 native 函数都是动态注册的&#xff0c;而且注册过程有固定的套路。我们以Parcel类为例来解析套路。 首先&#xff0c;我们在Parcel.java中会看到很多标记为…

Java-02 深入浅出 MyBatis - MyBatis 快速入门(无 Spring) POM Mapper 核心文件 增删改查

点一下关注吧&#xff01;&#xff01;&#xff01;非常感谢&#xff01;&#xff01;持续更新&#xff01;&#xff01;&#xff01; 大数据篇正在更新&#xff01;https://blog.csdn.net/w776341482/category_12713819.html 目前已经更新到了&#xff1a; MyBatis&#xff…

freemarker 读取template.xml ,通过response 输出文件,解决中文乱码问题

采用 try (Writer writer new OutputStreamWriter(os, “UTF-8”)) UTF-8 内容转换 public static void setResponseHeader(HttpServletResponse response, String fileName) {try {// fileName "中文.xls";try {fileName new String(fileName.getBytes(),"…

PostgreSQL中的COPY命令:高效数据导入与导出

在PostgreSQL数据库中&#xff0c;数据导入和导出是日常工作中常见的操作。传统的插入&#xff08;INSERT&#xff09;方法虽然可以实现数据的导入&#xff0c;但在处理大量数据时效率较低。而COPY命令则提供了一个快速、高效的方式来完成这一任务。COPY命令不仅可以用于将数据…

在启动 Spring Boot 项目时,报找不到 slf4j 的错误

而且 tomcat 的启动信息不知道为什么输出出来了 问 AI 得到的解决方案&#xff1a; 将 pom.xml 中的如下配置替换成这样&#xff0c;排除这个插件 <dependency><groupId>org.springframework.boot</groupId><artifactId>spring - boot - starter - …

论文阅读--supervised learning with quantum enhanced feature spaces

简略摘要 量子算法实现计算加速的核心要素是通过可控纠缠和干涉利用指数级大的量子态空间。本文在超导处理器上提出并实验实现了两种量子算法。这两种方法的一个关键组成部分是使用量子态空间作为特征空间。只有在量子计算机上才能有效访问的量子增强特征空间的使用为量子优势提…

【数字静态时序分析】复杂时钟树的时序约束SDC写法

以上图为例&#xff0c;SoC芯片上往往存在几种不同的时钟源&#xff0c;有pll时钟、环振时钟、外部的晶振时钟&#xff0c;在SoC不同的模块或者不同的运行阶段使用的时钟也往往不同&#xff0c;所以在使用的时候&#xff0c;相同的模块会出现选择不同的时钟源的情况。上图的情形…