【大数据】Flink 测试利器：DataGen

Flink 测试利器：DataGen

  • 1.什么是 FlinkSQL ？
  • 2.什么是 Connector ？
  • 3.DataGen Connector
    • 3.1 Demo
    • 3.2 支持的类型
    • 3.3 连接器属性
  • 4.DataGen 使用案例
    • 4.1 场景一：生成一亿条数据到 Hive 表
    • 4.2 场景二：持续每秒生产 10 万条数到消息队列
  • 5.思考

1.什么是 FlinkSQL ？

Flink SQL 是基于 Apache Calcite 的 SQL 解析器和优化器构建的，支持 ANSI SQL 标准，允许使用标准的 SQL 语句来处理流式和批处理数据。通过 Flink SQL，可以以声明式的方式描述数据处理逻辑，而无需编写显式的代码。使用 Flink SQL，可以执行各种数据操作，如 过滤、聚合、连接 和 转换 等。它还提供了 窗口操作、时间处理 和 复杂事件处理 等功能，以满足流式数据处理的需求。

Flink SQL 提供了许多扩展功能和语法，以适应 Flink 的流式和批处理引擎的特性。它是 Flink 最高级别的抽象，可以与 DataStream API 和 DataSet API 无缝集成，利用 Flink 的分布式计算能力和容错机制。

在这里插入图片描述
使用 Flink SQL 处理数据的基本步骤：

  • 定义输入表：使用 CREATE TABLE 语句定义输入表，指定表的模式（字段和类型）和数据源（如 Kafka、文件等）。
  • 执行 SQL 查询：使用 SELECT、INSERT INTO 等 SQL 语句来执行数据查询和操作。您可以在 SQL 查询中使用各种内置函数、聚合操作、窗口操作和时间属性等。
  • 定义输出表：使用 CREATE TABLE 语句定义输出表，指定表的模式和目标数据存储（如 Kafka、文件等）。
  • 提交作业：将 Flink SQL 查询作为 Flink 作业提交到 Flink 集群中执行。Flink 会根据查询的逻辑和配置自动构建执行计划，并将数据处理任务分发到集群中的任务管理器进行执行。

总而言之，我们可以通过 Flink SQL 查询和操作来处理流式和批处理数据。它提供了一种简化和加速数据处理开发的方式，尤其适用于熟悉 SQL 的开发人员和数据工程师。

2.什么是 Connector ？

Flink Connector 是指 用于连接外部系统和数据源的组件。它允许 Flink 通过特定的连接器与不同的数据源进行交互，例如数据库、消息队列、文件系统等。它负责处理与外部系统的通信、数据格式转换、数据读取和写入等任务。无论是作为输入数据表还是输出数据表，通过使用适当的连接器，可以在 Flink SQL 中访问和操作外部系统中的数据。目前实时平台提供了很多常用的连接器:

例如：

  • JDBC：用于与关系型数据库（如 MySQL、PostgreSQL）建立连接，并支持在 Flink SQL 中读取和写入数据库表的数据。
  • JDQ：用于与 JDQ 集成，可以读取和写入 JDQ 主题中的数据。
  • Elasticsearch：用于与 Elasticsearch 集成，可以将数据写入 Elasticsearch 索引或从索引中读取数据。
  • File Connector：用于读取和写入各种文件格式（如 CSV、JSON、Parquet）的数据。
  • ……

还有如 HBase、JMQ4、Doris、Clickhouse，Jimdb，Hive 等，用于与不同的数据源进行集成。通过使用 Flink SQL Connector，我们可以轻松地与外部系统进行数据交互，将数据导入到 Flink 进行处理，或 将处理结果导出到外部系统。

在这里插入图片描述

3.DataGen Connector

DataGen 是 Flink SQL 提供的一个内置连接器，用于生成模拟的测试数据，以便在开发和测试过程中使用。

使用 DataGen，可以生成具有不同数据类型和分布的数据，例如整数、字符串、日期等。这样可以模拟真实的数据场景，并帮助验证和调试 Flink SQL 查询和操作。

3.1 Demo

以下是一个使用 DataGen 函数的简单示例：

-- 创建输入表
CREATE TABLE input_table (order_number BIGINT,price DECIMAL(32,2),buyer ROW <first_name STRING,last_name STRING>,order_time TIMESTAMP(3)
) WITH ('connector' = 'datagen',
);

在上面的示例中，我们使用 DataGen 连接器创建了一个名为 input_table 的输入表。该表包含了 order_number、price、buyer、order_time 四个字段。默认是 Random 随机生成对应类型的数据，生产速率是 10000 10000 10000 条/秒，只要任务不停，就会源源不断的生产数据。当然也可以指定一些参数来定义生成数据的规则，例如每秒生成的行数、字段的数据类型和分布。

生成的数据样例：

{"order_number": -6353089831284155505,"price": 253422671148527900374700392448,"buyer": {"first_name": "6e4df4455bed12c8ad74f03471e5d8e3141d7977bcc5bef88a57102dac71ac9a9dbef00f406ce9bddaf3741f37330e5fb9d2","last_name": "d7d8a39e063fbd2beac91c791dc1024e2b1f0857b85990fbb5c4eac32445951aad0a2bcffd3a29b2a08b057a0b31aa689ed7"},"order_time": "2023-09-21 06:22:29.618"
}
{"order_number": 1102733628546646982,"price": 628524591222898424803263250432,"buyer": {"first_name": "4738f237436b70c80e504b95f0d9ec3d7c01c8745edf21495f17bb4d7044b4950943014f26b5d7fdaed10db37a632849b96c","last_name": "7f9dbdbed581b687989665b97c09dec1a617c830c048446bf31c746898e1bccfe21a5969ee174a1d69845be7163b5e375a09"},"order_time": "2023-09-21 06:23:01.69"
}

3.2 支持的类型

字段类型数据生成方式
BOOLEANrandom
CHARrandom / sequence
VARCHARrandom / sequence
STRINGrandom / sequence
DECIMALrandom / sequence
TINYINTrandom / sequence
SMALLINTrandom / sequence
INTrandom / sequence
BIGINTrandom / sequence
FLOATrandom / sequence
DOUBLErandom / sequence
DATErandom
TIMErandom
TIMESTAMPrandom
TIMESTAMP_LTZrandom
INTERVAL YEAR TO MONTHrandom
INTERVAL DAY TO MONTHrandom
ROWrandom
ARRAYrandom
MAPrandom
MULTISETrandom

3.3 连接器属性

属性是否必填默认值类型
描述
connectorrequired(none)String‘datagen’
rows-per-secondoptional 10000 10000 10000Long数据生产速率
number-of-rowsoptional(none)Long指定生产的数据条数，默认是不限制
fields.#.kindoptionalrandomString指定字段的生产数据的方式 random 还是 sequence
fields.#.minoptional(Minimum value of type)(Type of field)random 生成器的指定字段 # 最小值，支持数字类型
fields.#.maxoptional(Maximum value of type)(Type of field)random 生成器的指定字段 # 最大值，支持数字类型
fields.#.lengthoptional 100 100 100Integerchar / varchar / string / array / map / multiset 类型的长度
fields.#.startoptional(none)(Type of field)sequence 生成器的开始值
fields.#.endoptional(none)(Type of field)sequence 生成器的结束值

4.DataGen 使用案例

4.1 场景一：生成一亿条数据到 Hive 表

CREATE TABLE dataGenSourceTable (order_number BIGINT,price DECIMAL(10, 2),buyer STRING,order_time TIMESTAMP(3)
) WITH ( 'connector'='datagen', 'number-of-rows'='100000000','rows-per-second' = '100000'
);CREATE CATALOG myhive
WITH ('type'='hive','default-database'='default'
);
USE CATALOG myhive;
USE dev;
SET table.sql-dialect=hive;
CREATE TABLE if not exists shipu3_test_0932 (order_number BIGINT,price DECIMAL(10, 2),buyer STRING,order_time TIMESTAMP(3)
) PARTITIONED BY (dt STRING) STORED AS parquet TBLPROPERTIES ('partition.time-extractor.timestamp-pattern'='$dt','sink.partition-commit.trigger'='partition-time','sink.partition-commit.delay'='1 h','sink.partition-commit.policy.kind'='metastore,success-file'
);
SET table.sql-dialect=default;
insert into myhive.dev.shipu3_test_0932
select order_number, price, buyer, order_time, cast(CURRENT_DATE as varchar)
from default_catalog.default_database.dataGenSourceTable;

当每秒生产 10 万条数据的时候，17 分钟左右就可以完成，当然我们可以通过增加 Flink 任务的计算节点、并行度、提高生产速率 rows-per-second 的值等来更快速的完成大数据量的生产。

4.2 场景二：持续每秒生产 10 万条数到消息队列

CREATE TABLE dataGenSourceTable (order_number BIGINT,price INT,buyer ROW <first_name STRING,last_name STRING>,order_time TIMESTAMP(3),col_array ARRAY <STRING>,col_map map <STRING,STRING>
) WITH ( 'connector'='datagen', --连接器类型'rows-per-second'='100000', --生产速率'fields.order_number.kind'='random', --字段order_number的生产方式'fields.order_number.min'='1', --字段order_number最小值'fields.order_number.max'='1000', --字段order_number最大值'fields.price.kind'='sequence', --字段price的生产方式'fields.price.start'='1', --字段price开始值'fields.price.end'='1000', --字段price最大值'fields.col_array.element.length'='5', --每个元素的长度'fields.col_map.key.length'='5', --map key的长度'fields.col_map.value.length'='5' --map value的长度
);CREATE TABLE jdqsink1 (order_number BIGINT,price DECIMAL(32, 2),buyer ROW <first_name STRING,last_name STRING>,order_time TIMESTAMP(3),col_ARRAY ARRAY <STRING>,col_map map <STRING,STRING>
) WITH ('connector'='jdq','topic'='jrdw-fk-area_info__1','jdq.client.id'='xxxxx','jdq.password'='xxxxxxx','jdq.domain'='db.test.group.com','format'='json'
);INSERTINTO jdqsink1
SELECT * FROM dataGenSourceTable;

5.思考

通过以上案例可以看到，通过 Datagen 结合其他连接器可以模拟各种场景的数据。

  • 性能测试：我们可以利用 Flink 的高处理性能，来调试任务的外部依赖的阈值（超时，限流等）到一个合适的水位，避免自己的任务有过多的外部依赖出现木桶效应。
  • 边界条件测试：我们通过使用 Flink DataGen 生成特殊的测试数据，如最小值、最大值、空值、重复值等来验证 Flink 任务在边界条件下的正确性和鲁棒性。
  • 数据完整性测试：我们通过 Flink DataGen 可以生成包含错误或异常数据的数据集，如无效的数据格式、缺失的字段、重复的数据等。从而可以测试 Flink 任务对异常情况的处理能力，验证 Flink 任务在处理数据时是否能够正确地保持数据的完整性。

总之，Flink DataGen 是一个强大的工具，可以帮助测试人员构造各种类型的测试数据。通过合理的使用 ，测试人员可以更有效地进行测试，并发现潜在的问题和缺陷。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/413769.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

element-ui的el-upload组件实现上传拖拽排序图片顺序（sortablejs）

element-ui的el-upload组件实现上传拖拽排序图片顺序（sortablejs）

<template><!-- 省略其他配置 --><el-upload ref"upload" :file-list.sync"fileList"></el-upload></template><script>import Sortable from sortablejs;export default {data() {return {fileList: []};},mounted()…
阅读更多...
Cacti 前台SQL注入漏洞复现（CVE-2023-39361）

Cacti 前台SQL注入漏洞复现（CVE-2023-39361）

0x01 产品简介 Cacti 是一套基于 PHP,MySQL,SNMP 及 RRDTool 开发的网络流量监测图形分析工具。 0x02 漏洞概述 该漏洞存在于graph_view.php文件中。默认情况下，访客用户无需身份验证即可访问graph_view.php，在启用情况下使用时会导致SQL注入漏洞。 攻击者可能利用此漏洞…
阅读更多...
spark dateformat源码排错

spark dateformat源码排错

背景 有一个任务 yyyy写成了YYYY&#xff0c;导致年份不对触发告警 select from_unixtime(unix_timestamp(),YYYY-MM-dd HH:mm:ss) 第一时间用spark dateformat搜索下看看官网&#xff0c;发现spark 官网也没有描述YYYY的信息 Datetime patterns - Spark 3.5.0 Documentati…
阅读更多...
ELK日志分析

ELK日志分析

目录 1.ELK概述 ELK简述 ElasticSearch Kiabana Logstash 可添加其他的组件 Filebeat filebeat结合logstash带来好处 缓存/消息队列&#xff08;redis、kafka、RabbitMQ等&#xff09; Fluentd ELK的好处 完整日志系统基本特征 ELK的工作原理 2.ELK Elasticsearc…
阅读更多...
【计算机网络】OSI七层模型与TCP/IP四层模型的对应与各层介绍

【计算机网络】OSI七层模型与TCP/IP四层模型的对应与各层介绍

1 OSI七层模型与TCP/IP四层模型对应 2 OSI七层模型介绍 OSI&#xff08;Open Systems Interconnection&#xff09;模型是一个由国际标准化组织&#xff08;ISO&#xff09;定义的七层网络体系结构&#xff0c;用于描述计算机网络中的通信协议。每一层都有特定的功能&#xff…
阅读更多...
JOSEF约瑟 SS-17B时间继电器 柜内安装，板前接线带底座

JOSEF约瑟 SS-17B时间继电器 柜内安装，板前接线带底座

系列型号&#xff1a; SS-17B时间继电器&#xff1b; SS-17C时间继电器&#xff1b; SS-17C;SS-17B时间继电器 一、用途 SS-17系列时间继电器用于直流电路中&#xff0c;可在时间保护系统中作时间测量元件。适用于时间测量精度要求高&#xff0c;配合时间级差小 的场合&…
阅读更多...
web前端项目-中国象棋【附源码】

web前端项目-中国象棋【附源码】

中国象棋 【中国象棋】是一款历史悠久、深受人们喜爱的策略类游戏。在Web前端技术中&#xff0c;我们可以使用HTML、CSS和JavaScript等语言来制作一款中国象棋游戏。玩家使用棋子&#xff08;帅/相/士/炮/马/车/炮/卒&#xff09;在棋盘上相互对弈&#xff0c;将对手的“帅”棋…
阅读更多...
2024年回炉计划之排序算法（一）

2024年回炉计划之排序算法（一）

算法是计算机科学和信息技术中的重要领域&#xff0c;涉及到问题求解和数据处理的方法。要学习算法&#xff0c;你可能需要掌握以下一些基本知识&#xff1a; 基本数据结构&#xff1a; 了解和熟练使用各种数据结构&#xff0c;如数组、链表、栈、队列、树和图等。数据结构是算…
阅读更多...
深度学习中Numpy的一些注意点（多维数组；数据类型转换、数组扁平化、np.where()、np.argmax()、图像拼接、生成同shape的图片）

深度学习中Numpy的一些注意点（多维数组；数据类型转换、数组扁平化、np.where()、np.argmax()、图像拼接、生成同shape的图片）

文章目录 1多维数组压缩维度扩充维度 2numpy类型转换深度学习常见的float32类型。 3数组扁平化4np.where()的用法5np.argmax()6图像拼接7生成同shape的图片&#xff0c;指定数据类型 1多维数组 a.shape(3,2);既数组h3&#xff0c;w2 a.shape(2,3,2);这里第一个2表示axis0维度上…
阅读更多...
Docker容器添加映射端口

Docker容器添加映射端口

方式一 简单粗暴&#xff08;需要等一段时间&#xff09; 直接给现在容器停了&#xff08;当然你要不想停也可以&#xff0c;只是打包会慢一点&#xff0c;当然我是没出意外&#xff0c;如果你怕出现特殊情况&#xff0c;那就先把容器停了&#xff09;&#xff0c;然后把这个容…
阅读更多...
SQL Povit函数使用及实例

SQL Povit函数使用及实例

PIVOT函数常用于数据的行转列&#xff0c;同时也可以用此函数实现类似于Excel中的数据透视表的效果。 PIVOT函数 PIVOT 函数的基本语法如下&#xff1a; -- PIVOT 语法 SELECT <非透视的列>,[第一个透视的列] AS <列名称>,[第二个透视的列] AS <列名称>,.…
阅读更多...
【springboot】配置文件入门

【springboot】配置文件入门

配置文件入门 配置文件最重要的目的&#xff1a;解决硬编码问题(代码写死) 我们接下来主要介绍两个方面&#xff1a;常见的配置项和配置文件的使用 SpringBoot 的配置文件,有三种格式 propertiesyamlyml(yaml的简写) 用的较多的是yml和properties文件 如果项目中,同时存在…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023