文档首页/
MapReduce服务 MRS/
组件操作指南(LTS版)/
使用Flink/
Flink企业级能力增强/
FlinkSQL Kafka和upsert-kafka Connector支持限流读
更新时间:2025-08-01 GMT+08:00
FlinkSQL Kafka和upsert-kafka Connector支持限流读
本章节适用于MRS 3.3.0及以后版本。
使用场景
使用FlinkSQL的Kafka、upsert-kafka Connector消费数据时需要限流。
使用方法
在创建的Source流表中添加“subtask.scan.records-per-second.limit”参数,该参数表示每秒消费Kafka单分区记录数,因此Source端整体限流速率为:min( source parallelism * subtask.scan.records-per-second.limit,kafka partitions num * subtask.scan.records-per-second.limit)。
SQL示例如下:
CREATE TABLE KafkaSource ( `user_id` VARCHAR, `user_name` VARCHAR, `age` INT ) WITH ( 'connector' = 'kafka', 'topic' = 'test_source', 'properties.bootstrap.servers' = 'Kafka的Broker实例业务IP:Kafka端口号', 'properties.group.id' = 'testGroup', 'scan.startup.mode' = 'latest-offset', 'format' = 'csv', 'subtask.scan.records-per-second.limit' = '1000', --表示每秒消费1000条记录 'properties.sasl.kerberos.service.name' = 'kafka', 'properties.security.protocol' = 'SASL_PLAINTEXT', 'properties.kerberos.domain.name' = 'hadoop.系统域名' ); CREATE TABLE KafkaSink( `user_id` VARCHAR, `user_name` VARCHAR, `age` INT ) WITH ( 'connector' = 'kafka', 'topic' = 'test_sink', 'properties.bootstrap.servers' = 'Kafka的Broker实例业务IP:Kafka端口号', 'scan.startup.mode' = 'latest-offset', 'format' = 'csv', 'properties.sasl.kerberos.service.name' = 'kafka', 'properties.security.protocol' = 'SASL_PLAINTEXT', 'properties.kerberos.domain.name' = 'hadoop.系统域名' ); Insert into KafkaSink select * from KafkaSource;

- Kafka Broker实例IP地址及端口号说明:
- 服务的实例IP地址可通过登录FusionInsight Manager后,单击“集群 > 服务 > Kafka > 实例”,在实例列表页面中查询。
- 集群已启用Kerberos认证(安全模式)时Broker端口为“sasl.port”参数的值,默认为“21007”。
- 集群未启用Kerberos认证(普通模式)时Broker端口为“port”的值,默认为“9092”。如果配置端口号为9092,则需要配置“allow.everyone.if.no.acl.found”参数为true,具体操作如下:
登录FusionInsight Manager系统,选择“集群 > 服务 > Kafka > 配置 > 全部配置”,搜索“allow.everyone.if.no.acl.found”配置,修改参数值为true,保存配置即可。
- 系统域名:可登录FusionInsight Manager,选择“系统 > 权限 > 域和互信”,查看“本端域”参数,即为当前系统域名。
父主题: Flink企业级能力增强