学习笔记Flink（七）—— Flink Kafka插件

添加依赖& API

在pom.xml添加：

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka_2.11</artifactId><version>1.10.1</version>
</dependency>

代码：

package flink_kafkaimport java.util.Properties
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerobject MyFlinkKafkaConsumer {def main(args: Array[String]): Unit = {val properties = new Properties()properties.put("bootstrap.servers", "node110:9092,node111:9092,node112:9092")properties.put("group.id", "test")val env = StreamExecutionEnvironment.getExecutionEnvironment//create kafka sourceval kafkaSource = env.addSource(new FlinkKafkaConsumer[String]("demo02",//topicnew SimpleStringSchema(),//seriableproperties//kafka cluster configuration))//SinkkafkaSource.print()//executeenv.execute("read from kafka demo02")}
}

运行测试：
① 创建demo02话题，并在demo02写入数据

② 执行代码

Flink作为输出

package flink_kafkaimport java.util.Propertiesimport org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.connectors.kafka.internals.KeyedSerializationSchemaWrapper
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer, FlinkKafkaProducer}object MyFlinkKafkaConsumerAndProducer {def main(args: Array[String]): Unit = {val properties = new Properties()properties.put("bootstrap.servers", "node110:9092,node111:9092,node112:9092")properties.put("group.id", "test")val env = StreamExecutionEnvironment.getExecutionEnvironment//create kafka sourceval kafkaSource = env.addSource(new FlinkKafkaConsumer[String]("demo02",//topicnew SimpleStringSchema(),//seriableproperties//kafka cluster configuration))//transformationval processed = kafkaSource.flatMap(_.split("\\w+")).map((_,1)).keyBy(0).timeWindow(Time.seconds(5)).sum(1).filter(_._2>=3).map(row => row._1+"->"+row._2)//kafka Sinkval kafkaProducer = new FlinkKafkaProducer[String]("demo01",//target topicnew KeyedSerializationSchemaWrapper[String](new SimpleStringSchema()),//seriablization schemaproperties,FlinkKafkaProducer.Semantic.EXACTLY_ONCE)processed.addSink(kafkaProducer)//executeenv.execute("read from kafka demo02 and write to demo01")}
}

测试：

学习笔记Flink（七）—— Flink Kafka插件相关推荐

七天入门linux,RHCE认证学习笔记-第七天
RHCE认证学习笔记-第七天 1.系统启动过程: BIOS初始化-->引导程序(Bootloader)-->内核初始化-->进程(Init)-->用户程序 (1)BIOS初始化 ...
Intel VT学习笔记（七）—— EPT物理地址转换
Intel VT学习笔记(七)-- EPT物理地址转换要点回顾 EPT 支持检测 9-9-9-9-12分页实验:EPT物理地址转换参考资料要点回顾在上一篇中,已经初步实现了最小VT框架,但实 ...
软件调试学习笔记（七）—— 单步步入单步步过
软件调试学习笔记(七)-- 单步步入&单步步过单步步入设置单步异常处理单步异常实验1:单步异常的设置与处理单步步过实现思路实验2:实现单步步过单步步入描述: 单步步入的实现依 ...
Windows驱动开发学习笔记（七）—— 多核同步内核重载
Windows驱动开发学习笔记(七)-- 多核同步基础知识并发与同步分析 InterlockedIncrement 原子操作相关API 内核文件多核同步临界区示例一:错误的临界区示例二: ...
Windows进程与线程学习笔记（七）—— 时间片管理
Windows进程与线程学习笔记(七)-- 时间片管理要点回顾基本概念 CPU时间片分析 KeUpdateRunTime 分析 KiDispatchInterrupt 备用线程总结要点回顾 ...
Windows保护模式学习笔记（七）—— PDEPTE
Windows保护模式学习笔记(七)-- PDE&PTE Cr3 PDE(页目录表项) PTE(页表项) 物理页的属性 10-10-12分页的补充实验1:证明PTE的特征1 第一步:选择一个 ...
JavaScript学习笔记（七）——厚积薄发之小成果
JavaScript学习笔记(七)--厚积薄发之小成果目前我先列好提纲,利用每晚和周末的闲暇时间,将逐步写完 ^_^ 转载于:https://www.cnblogs.com/wdpp/archive ...
Mysql学习笔记（七）查(补充)
Mysql学习笔记(七)查(补充) 原文:Mysql学习笔记(七)查(补充) PS:五一还是要学习...虽然有点苦逼..但是路是自己选的,那么自己就要坚持的走下去... 学习内容: 1.数据库查找的补 ...
Kafka学习笔记（3）----Kafka的数据复制（Replica）与Failover
1. CAP理论 1.1 Cosistency(一致性) 通过某个节点的写操作结果对后面通过其他节点的读操作可见. 如果更新数据后,并发访问的情况下可立即感知该更新,称为强一致性如果允许之后部分或全 ...
OpenCV学习笔记（七）：形态学morpholgy(1)：腐蚀/膨胀：enrode(),dilate()
OpenCV学习笔记(七):形态学(morpholgy):腐蚀/膨胀:enrode(),dilate() 数学形态学(Mathematical morphology) 是一门建立在格论和拓扑学基础之上 ...

学习笔记Flink（七）—— Flink Kafka插件

添加依赖& API

Flink作为输出

学习笔记Flink（七）—— Flink Kafka插件相关推荐

最新文章

热门文章