Flink Sink Kafka
时间:2021-08-16
本文章向大家介绍Flink Sink Kafka,主要包括Flink Sink Kafka使用实例、应用技巧、基本知识点总结和需要注意事项,具有一定的参考价值,需要的朋友可以参考一下。
原创转载请注明出处:https://www.cnblogs.com/agilestyle/p/15149998.html
Kafka Setup
wget https://downloads.apache.org/kafka/2.8.0/kafka_2.12-2.8.0.tgz tar zxvf kafka_2.12-2.8.0.tgz -C ~/app cd ~/app/kafka_2.12-2.8.0 # start the zookeeper service bin/zookeeper-server-start.sh config/zookeeper.properties # start the Kafka broker service bin/kafka-server-start.sh config/server.properties # create topic bin/kafka-topics.sh --create --topic flink_topic --bootstrap-server localhost:9092 # describle message bin/kafka-topics.sh --describe --topic flink_topic --bootstrap-server localhost:9092 # produce message bin/kafka-console-producer.sh --topic flink_topic --bootstrap-server localhost:9092 # consumer message bin/kafka-console-consumer.sh --topic flink_topic --from-beginning --bootstrap-server localhost:9092
Maven Dependency
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>org.fool</groupId> <artifactId>flink</artifactId> <version>1.0-SNAPSHOT</version> <properties> <maven.compiler.source>8</maven.compiler.source> <maven.compiler.target>8</maven.compiler.target> </properties> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>1.12.5</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java_2.12</artifactId> <version>1.12.5</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients_2.12</artifactId> <version>1.12.5</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka_2.12</artifactId> <version>1.12.1</version> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <version>1.18.20</version> </dependency> </dependencies> </project>
SRC
src/main/java/org/fool/flink/contract/Sensor.java
package org.fool.flink.contract; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; @Data @NoArgsConstructor @AllArgsConstructor public class Sensor { private String id; private Long timestamp; private Double temperature; }
src/main/java/org/fool/flink/sink/SinkKafkaTest.java
package org.fool.flink.sink; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer; import org.fool.flink.contract.Sensor; import java.util.Objects; import java.util.Properties; public class SinkKafkaTest { public static void main(String[] args) throws Exception { StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment(); environment.setParallelism(1); Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "localhost:9092"); properties.setProperty("group.id", "consumer-group"); properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); properties.setProperty("auto.offset.reset", "latest"); String inputPath = Objects.requireNonNull(ClassLoader.getSystemClassLoader().getResource("sensor.txt")).getPath(); DataStream<String> inputStream = environment.readTextFile(inputPath); DataStream<String> dataStream = inputStream.map(new MapFunction<String, String>() { @Override public String map(String value) throws Exception { String[] fields = value.split(","); return new Sensor(fields[0], new Long(fields[1]), new Double(fields[2])).toString(); } }); dataStream.addSink(new FlinkKafkaProducer<>("flink_topic", new SimpleStringSchema(), properties)); environment.execute(); } }
src/main/resources/sensor.txt
1,1628754086,35.8 2,1628754096,36.8 3,1628754106,37.8 4,1628754116,38.8 1,1628754186,36.6 2,1628754296,36.6 3,1628754306,37.6 4,1628754416,38.6 1,1628754986,25.8 1,1628754086,39.6 2,1628754996,26.8 3,1628754906,27.8 4,1628754916,28.8
Run
Console Output
Kafka Consumer
欢迎点赞关注和收藏
强者自救 圣者渡人
原文地址:https://www.cnblogs.com/agilestyle/p/15149998.html
- TensorFlow从1到2 - 5 - 非专家莫入!TensorFlow实现CNN
- JetBrains Rider 破解 (ideaIU等等开发工具都通用)
- python中的小魔法(一)
- 由问题入手,步步爬出Python中赋值与拷贝的坑
- python爬取链家租房之获得每一页的房屋信息地址(持续更新)
- python使用正则表达式
- python在租房过程中的应用
- python爬虫反爬取---设置IP代理自动变换requests.get()中proxy的IP
- 【译】TensorFlow实现Batch Normalization
- 关于Python语言规范你需要知道的一些小tips
- R语言可视化——REmap(路径图)
- python面向对象
- 字符串hash入门
- R语言数据处理——数据合并与追加
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- jmeter 在linux服务器的安装和运行教程图解
- Linux中改变文件权限的chmod命令详析
- Ubuntu常用指令及用法详解
- Linux 系统使用私钥登录的教程详解
- Ubuntu如何轻松编译openJDK详解
- 详解Linux下Tomcat开启查看GC信息
- Centos7.4 zabbix3.4.7源码安装的方法步骤
- CentOS7环境搭建python3以及与python2实现共存的方法
- centos7.0安装离线JDK1.8方法详解
- ubuntu中swap(虚拟内存)设置方法
- 详解Linux监控重要进程的实现方法
- CentOS环境下安装Redis3.0及phpredis扩展测试示例
- 使用Apache commons-cli包进行命令行参数解析的示例代码
- 详解如何在Ubuntu 16.04上增加Swap分区
- Mac本地文件上传到CentOS云服务器方法