Flink KafkaSink

Kafka集成

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

方案1

import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema
import org.apache.kafka.clients.producer.ProducerRecord
class UserDefinedKafkaSerializationSchema extends
KafkaSerializationSchema[(String,Int)]{override def serialize(element: (String, Int), timestamp: lang.Long):
ProducerRecord[Array[Byte], Array[Byte]] = {return new
ProducerRecord("topic01",element._1.getBytes(),element._2.toString.getBytes())}
}

//1.创建流计算执⾏环境
val env = StreamExecutionEnvironment.getExecutionEnvironmentenv.setParallelism(4)//2.创建DataStream - 细化val text = env.readTextFile("hdfs://CentOS:9000/demo/words")val props = new Properties()props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "CentOS:9092")props.setProperty(ProducerConfig.BATCH_SIZE_CONFIG,"100")props.setProperty(ProducerConfig.LINGER_MS_CONFIG,"500")//Semantic.EXACTLY_ONCE:开启kafka幂等写特性//Semantic.AT_LEAST_ONCE:开启Kafka Retries机制val kafakaSink = new FlinkKafkaProducer[(String, Int)]("defult_topic",new
UserDefinedKafkaSerializationSchema, props,Semantic.AT_LEAST_ONCE)//3.执⾏DataStream的转换算⼦val counts = text.flatMap(line=>line.split("\\s+")).map(word=>(word,1)).keyBy(0).sum(1)counts.addSink(kafakaSink)//5.执⾏流计算任务env.execute("Window Stream WordCount")

以上的defult_topic没有任何意义

方案2–老版

class UserDefinedKeyedSerializationSchema extends
KeyedSerializationSchema[(String,Int)]{Intoverride def serializeKey(element: (String, Int)): Array[Byte] = {element._1.getBytes()}override def serializeValue(element: (String, Int)): Array[Byte] = {element._2.toString.getBytes()}//可以覆盖 默认是topic，如果返回null，则将数据写⼊到默认的topic中override def getTargetTopic(element: (String, Int)): String = {null}
}

//1.创建流计算执⾏环境
val env = StreamExecutionEnvironment.getExecutionEnvironmentenv.setParallelism(4)//2.创建DataStream - 细化val text = env.readTextFile("hdfs://CentOS:9000/demo/words")
Operators
DataStream Transformations
DataStream → DataStream
Map
Takes one element and produces one element. A map function that doubles the values of the input stream:
FlatMap
Takes one element and produces zero, one, or more elements. A flatmap function that splits sentences to
words:
Filter
Evaluates a boolean function for each element and retains those for which the function returns true. A filter
that filters out zero values:
DataStream* → DataStreamval props = new Properties()props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "CentOS:9092")props.setProperty(ProducerConfig.BATCH_SIZE_CONFIG,"100")props.setProperty(ProducerConfig.LINGER_MS_CONFIG,"500")//Semantic.EXACTLY_ONCE:开启kafka幂等写特性//Semantic.AT_LEAST_ONCE:开启Kafka Retries机制val kafakaSink = new FlinkKafkaProducer[(String, Int)]("defult_topic",new
UserDefinedKeyedSerializationSchema, props, Semantic.AT_LEAST_ONCE)//3.执⾏DataStream的转换算⼦val counts = text.flatMap(line=>line.split("\\s+")).map(word=>(word,1)).keyBy(0).sum(1)counts.addSink(kafakaSink)//5.执⾏流计算任务env.execute("Window Stream WordCount")

Flink KafkaSink相关推荐

2021年大数据Flink（四十四）：扩展阅读 End-to-End Exactly-Once
目录扩展阅读 End-to-End Exactly-Once 流处理的数据处理语义 At-most-once-最多一次 At-least-once-至少一次 Exactly-once-精确一次 En ...
2021年大数据Flink（二十七）：Flink 容错机制 Checkpoint
目录 Flink 容错机制 Checkpoint State Vs Checkpoint Checkpoint执行流程简单流程复杂流程 State状态后端/State存储介质 MemStateBa ...
2021年大数据Flink（十五）：流批一体API Connectors Kafka
目录 Kafka pom依赖参数设置参数说明 Kafka命令代码实现-Kafka Consumer 代码实现-Kafka Producer 代码实现-实时ETL Kafka pom依赖 Flin ...
Flink or Spark?实时计算框架在K12场景的应用实践
如今,越来越多的业务场景要求 OLTP 系统能及时得到业务数据计算.分析后的结果,这就需要实时的流式计算如Flink等来保障.例如,在 TB 级别数据量的数据库中,通过 SQL 语句或相关 API直接 ...
技术实践 | 如何基于 Flink 实现通用的聚合指标计算框架
导读:网易云信作为一个 PaaS 服务,需要对线上业务进行实时监控,实时感知服务的"心跳"."脉搏"."血压"等健康状况.通过采集服务拿到 ...
1.12.Flink Kafka-Connector详解、Consumer消费策略设置、动态加载Topic、Consumers Offset 自动提交、Producer、容错等
1.12.Flink Kafka-Connector详解 1.12.1.Kafka Consumer消费策略设置 1.12.2.Kafka Consumer的容错 1.12.3.动态加载Topic 1 ...
Apache Flink 零基础入门（二十）Flink kafka connector
内置source和sink 内置source包括从文件读取,从文件夹读取,从socket中读取.从集合或者迭代器中读取.内置的sink包括写文件.控制台输出.socket 内置connectors A ...
【3】flink sink
[README] 本文记录了flink sink操作,输出目的存储器(中间件)包括 kafka: es: db: 等等有很多: 本文只给出了 sink2kafka的代码: 本文使用的flink为 1. ...
flink API之Sink入门
kafka sink 添加依赖 <dependency><groupId>org.apache.flink</groupId><artifactId>f ...

Flink KafkaSink

Kafka集成

Flink KafkaSink相关推荐

最新文章

热门文章