本文为您介绍如何使用Datastream作业读取阿里云Kafka数据。
前提条件
- 本地安装了Java JDK 8。
- 本地安装了Maven 3.x。
- 本地安装了用于Java或Scala开发的IDE,推荐IntelliJ IDEA,且已配置完成JDK和Maven环境。
- 创建与实时计算独享模式相同VPC的Kafka实例,并创建了Topic和Consumer Group。
背景信息
- 实时计算Datastream完全兼容开源Flink 1.5.2版本,阿里云Kafka兼容开源Kafka,因此可以直接使用Maven仓库里的Kafka Connetor来连接阿里云Kafka。
- 实时计算独享模式通过内网接入Kafka,无需进行SASL认证鉴权。如果您在本地IDE上通过公网方式接入阿里云Kafka,则需要进行SASL认证鉴权,具体配置请参见kafka-java-demo。
注意 仅Blink3.x版本支持本示例。
开发
上线
请参见上线完成作业上线。
本示例对应的作业内容如下。
说明 注意修改blink.main.class、blink.job.name和blink.main.jar。
--完整主类名,必填,例如com.alibaba.realtimecompute.DatastreamExample。
blink.main.class=com.alibaba.blink.datastreaming.AliKafkaConsumerDemo
--作业名称。
blink.job.name=alikafkaconsumerdemo
--包含完整主类名的JAR包资源名称,多个JAR包时必填,例如blink_datastream.jar。
blink.main.jar=blink-datastreaming-1.0-snapshot.jar
--默认statebackend配置,当作业代码没有显式配置时生效。
state.backend.type=niagara
state.backend.niagara.ttl.ms=129600000
--默认checkpoint配置,当作业代码没有显式配置时生效。
blink.checkpoint.interval.ms=180000
说明 您可以设置自定义参数,详情请参见自定义参数。
在文档使用中是否遇到以下问题
更多建议
匿名提交