flink API之Sink入门

kafka sink

添加依赖

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka-0.11_2.11</artifactId><version>1.7.2</version>
</dependency>

将数据写入kafka,然后启动命令行消费者可以看下结果

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011import java.util.Propertiesobject KafkaSink extends App {val properties = new Properties()properties.setProperty("bootstrap.servers", "mypc01:9092,mypc02:9092,mypc03: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")//构建环境private val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//读取文件private val stream: DataStream[String] = env.readTextFile("a.txt")//添加kafka sinkstream.addSink(new FlinkKafkaProducer011[String]("flink", new SimpleStringSchema(), properties))//执行env.execute()
}

在kafka集群上启动消费者查看结果即可

kafka-console-consumer.sh \
--bootstrap-server mypc01:9092,mypc02:9092,mypc03:9092 \
--topic flink

Redis Sink

<dependency><groupId>org.apache.bahir</groupId><artifactId>flink-connector-redis_2.11</artifactId><version>1.0</version>
</dependency>

flink API之Sink入门相关推荐

Flink API之Source入门
从集合构建 import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}object Sen ...
9.FLINK Sink\API\自定义sink
9.Sink 9.1.API 9.2.注意 9.3.自定义sink 9.Sink 9.1.API 1.ds.print 直接输出到控制台 2.ds.printToErr()直接输出到控制台,用红色 3 ...
Flink之流处理API之Sink
Sink Flink没有类似于spark中foreach方法,让用户进行迭代的操作.虽有对外的输出操作都要利用Sink完成.最后通过类似如下方式完成整个任务最终输出操作. myDstream.addS ...
flink Table API 与SQL入门实战
流处理和批处理都可以用,是非常的方便! 导入依赖 <dependency><groupId>org.apache.flink</groupId><artifa ...
Flink 教程 gitbook 从入门到入土（详细教程）
Flink从入门到入土(详细教程) 和其他所有的计算框架一样,flink也有一些基础的开发步骤以及基础,核心的API,从开发步骤的角度来讲,主要分为四大部分 1.Environment Flink J ...
Flink SQL 自定义 Sink
1. 背景 2. 步骤 3.自定义 sink 代码 4. 使用 Redis Sink 5.详细解释 6.原理 7.参考 1.背景内部要做 Flink SQL 平台,本文以自定义 Redis Sink ...
Flink实操 : Sink操作
. 一 .前言二 .类型 2.1. 基于本地集合的sink 2.2. 基于文件的sink 2.2.1.将数据写入本地文件 2.2.2.将数据写入HDFS 2.3. Kafka Sink 2.4. M ...
【Flink】Flink 自定义 redis sink
1.概述内部要做 Flink SQL 平台,本文以自定义 Redis Sink 为例来说明 Flink SQL 如何自定义 Sink 以及自定义完了之后如何使用基于 Flink 1.11 2.步骤 ...
初识Django —Python API接口编程入门
初识Django -Python API接口编程入门一.WEB架构的简单介绍 Django是什么? Django是一个开放源代码的Web应用框架,由Python写成.我们的目标是用Python语言, ...

flink API之Sink入门

kafka sink

Redis Sink

flink API之Sink入门相关推荐

最新文章

热门文章