Flink的sink实战之二:kafka

Flink的sink实战之二:kafka

欢迎访问我的GitHub

https://github.com/zq2599/blog_demos

内容:所有原创文章分类汇总及配套源码,涉及Java、Docker、Kubernetes、DevOPS等;

本篇概览

本文是《Flink的sink实战》系列的第二篇,前文《Flink的sink实战之一:初探》对sink有了基本的了解,本章来体验将数据sink到kafka的操作;

全系列链接

  1. 《Flink的sink实战之一:初探》
  2. 《Flink的sink实战之二:kafka》
  3. 《Flink的sink实战之三:cassandra3》
  4. 《Flink的sink实战之四:自定义》

版本和环境准备

本次实战的环境和版本如下:

  1. JDK:1.8.0_211
  2. Flink:1.9.2
  3. Maven:3.6.0
  4. 操作系统:macOS Catalina 10.15.3 (MacBook Pro 13-inch, 2018)
  5. IDEA:2018.3.5 (Ultimate Edition)
  6. Kafka:2.4.0
  7. Zookeeper:3.5.5

请确保上述环境和服务已经就绪;

源码下载

如果您不想写代码,整个系列的源码可在GitHub下载到,地址和链接信息如下表所示(https://github.com/zq2599/blog_demos):

名称 链接 备注
项目主页 https://github.com/zq2599/blog_demos 该项目在GitHub上的主页
git仓库地址(https) https://github.com/zq2599/blog_demos.git 该项目源码的仓库地址,https协议
git仓库地址(ssh) git@github.com:zq2599/blog_demos.git 该项目源码的仓库地址,ssh协议

这个git项目中有多个文件夹,本章的应用在flinksinkdemo文件夹下,如下图红框所示:
在这里插入图片描述
准备完毕,开始开发;

准备工作

正式编码前,先去官网查看相关资料了解基本情况:

  1. 地址:https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/connectors/kafka.html
  2. 我这里用的kafka是2.4.0版本,在官方文档查找对应的库和类,如下图红框所示:
    在这里插入图片描述

kafka准备

  1. 创建名为test006的topic,有四个分区,参考命令:
./kafka-topics.sh \
--create \
--bootstrap-server 127.0.0.1:9092 \
--replication-factor 1 \
--partitions 4 \
--topic test006
  1. 在控制台消费test006的消息,参考命令:
./kafka-console-consumer.sh \
--bootstrap-server 127.0.0.1:9092 \
--topic test006
  1. 此时如果该topic有消息进来,就会在控制台输出;
  2. 接下来开始编码;

创建工程

  1. 用maven命令创建flink工程:
mvn \
archetype:generate \
-DarchetypeGroupId=org.apache.flink \
-DarchetypeArtifactId=flink-quickstart-java \
-DarchetypeVersion=1.9.2
  1. 根据提示,groupid输入com.bolingcavalry,artifactid输入flinksinkdemo,即可创建一个maven工程;
  2. 在pom.xml中增加kafka依赖库:
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-kafka_2.11</artifactId>
  <version>1.9.0</version>
</dependency>
  1. 工程创建完成,开始编写flink任务的代码;

发送字符串消息的sink

先尝试发送字符串类型的消息:

  1. 创建KafkaSerializationSchema接口的实现类,后面这个类要作为创建sink对象的参数使用:
package com.bolingcavalry.addsink;

import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.nio.charset.StandardCharsets;

public class ProducerStringSerializationSchema implements KafkaSerializationSchema<String> {

    private String topic;

    public ProducerStringSerializationSchema(String topic) {
        super();
        this.topic = topic;
    }

    @Override
    public ProducerRecord<byte[], byte[]> serialize(String element, Long timestamp) {
        return new ProducerRecord<byte[], byte[]>(topic, element.getBytes(StandardCharsets.UTF_8));
    }
}
  1. 创建任务类KafkaStrSink,请注意FlinkKafkaProducer对象的参数,FlinkKafkaProducer.Semantic.EXACTLY_ONCE表示严格一次:
package com.bolingcavalry.addsink;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

public class KafkaStrSink {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //并行度为1
        env.setParallelism(1);

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.50.43:9092");

        String topic = "test006";
        FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>(topic,
                new ProducerStringSerializationSchema(topic),
                properties,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

        //创建一个List,里面有两个Tuple2元素
        List<String> list = new ArrayList<>();
        list.add("aaa");
        list.add("bbb");
        list.add("ccc");
        list.add("ddd");
        list.add("eee");
        list.add("fff");
        list.add("aaa");

        //统计每个单词的数量
        env.fromCollection(list)
           .addSink(producer)
           .setParallelism(4);

        env.execute("sink demo : kafka str");
    }
}
  1. 使用mvn命令编译构建,在target目录得到文件flinksinkdemo-1.0-SNAPSHOT.jar
  2. 在flink的web页面提交flinksinkdemo-1.0-SNAPSHOT.jar,并制定执行类,如下图:
    在这里插入图片描述
  3. 提交成功后,如果flink有四个可用slot,任务会立即执行,会在消费kafak消息的终端收到消息,如下图:
    在这里插入图片描述
  4. 任务执行情况如下图:
    在这里插入图片描述

发送对象消息的sink

再来尝试如何发送对象类型的消息,这里的对象选择常用的Tuple2对象:

  1. 创建KafkaSerializationSchema接口的实现类,该类后面要用作sink对象的入参,请注意代码中捕获异常的那段注释:生产环境慎用printStackTrace()!!!
package com.bolingcavalry.addsink;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;
import javax.annotation.Nullable;

public class ObjSerializationSchema implements KafkaSerializationSchema<Tuple2<String, Integer>> {

    private String topic;
    private ObjectMapper mapper;

    public ObjSerializationSchema(String topic) {
        super();
        this.topic = topic;
    }

    @Override
    public ProducerRecord<byte[], byte[]> serialize(Tuple2<String, Integer> stringIntegerTuple2, @Nullable Long timestamp) {
        byte[] b = null;
        if (mapper == null) {
            mapper = new ObjectMapper();
        }
        try {
            b= mapper.writeValueAsBytes(stringIntegerTuple2);
        } catch (JsonProcessingException e) {
            // 注意,在生产环境这是个非常危险的操作,
            // 过多的错误打印会严重影响系统性能,请根据生产环境情况做调整
            e.printStackTrace();
        }
        return new ProducerRecord<byte[], byte[]>(topic, b);
    }
}
  1. 创建flink任务类:
package com.bolingcavalry.addsink;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

public class KafkaObjSink {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //并行度为1
        env.setParallelism(1);

        Properties properties = new Properties();
        //kafka的broker地址
        properties.setProperty("bootstrap.servers", "192.168.50.43:9092");

        String topic = "test006";
        FlinkKafkaProducer<Tuple2<String, Integer>> producer = new FlinkKafkaProducer<>(topic,
                new ObjSerializationSchema(topic),
                properties,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

        //创建一个List,里面有两个Tuple2元素
        List<Tuple2<String, Integer>> list = new ArrayList<>();
        list.add(new Tuple2("aaa", 1));
        list.add(new Tuple2("bbb", 1));
        list.add(new Tuple2("ccc", 1));
        list.add(new Tuple2("ddd", 1));
        list.add(new Tuple2("eee", 1));
        list.add(new Tuple2("fff", 1));
        list.add(new Tuple2("aaa", 1));

        //统计每个单词的数量
        env.fromCollection(list)
            .keyBy(0)
            .sum(1)
            .addSink(producer)
            .setParallelism(4);
        
        env.execute("sink demo : kafka obj");
    }
}
  1. 像前一个任务那样编译构建,把jar提交到flink,并指定执行类是com.bolingcavalry.addsink.KafkaObjSink
  2. 消费kafka消息的控制台输出如下:
    在这里插入图片描述
  3. 在web页面可见执行情况如下:
    在这里插入图片描述
    至此,flink将计算结果作为kafka消息发送出去的实战就完成了,希望能给您提供参考,接下来的章节,我们会继续体验官方提供的sink能力;

欢迎关注公众号:程序员欣宸

微信搜索「程序员欣宸」,我是欣宸,期待与您一同畅游Java世界…
https://github.com/zq2599/blog_demos

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/2632.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • 怎么解决生理需求啊_activate pycharm什么意思

    怎么解决生理需求啊_activate pycharm什么意思最近在一直学习pythonDjangoweb开发,安装了好久的pycharm打不开了,显示如下:

  • Win8下用DOSBox编写汇编语言

    Win8下用DOSBox编写汇编语言      DOSBox 是一个 DOS模拟程序,可以很方便的移植到其他的平台,因此可以使用它在Win8系统中编写汇编语言。下面提供相关下载链接:     DOSBox下载: http://pan.baidu.com/s/1qWwkss0      汇编工具下载: http://pan.baidu.com/s/1i3taKmT  DOSB

  • IT学生学习指南

    IT学生学习指南本文主要为了计算机相关专业的学生而写。内容是从本博主的工作经验和教学经验当中总结出来的一些干货,涵盖了从大一到大四期间学生所遇到的常见问题,包括如何找到满意的工作、如何选择适合自己的工作岗位、如何制定自己的学习计划、如何分辨自身适合考研还是就业、女生的IT之路怎么走、如何锻炼自己的技术等等大学生最关心的问题。如果大家觉得文本值得一看,那就请推荐给身边那些还在迷茫中的孩子们吧!

  • 显示搜索dota2协调服务器,老司机教你处理搜索dota2游戏协调服务器中【操作流程】…[通俗易懂]

    显示搜索dota2协调服务器,老司机教你处理搜索dota2游戏协调服务器中【操作流程】…[通俗易懂]win7系统有很多人都喜欢使用,我们操作的过程中常常会碰到win7系统搜索dota2游戏协调服务器中的问题。如果遇到win7系统搜索dota2游戏协调服务器中的问题该怎么办呢?很多电脑水平薄弱的网友不知道win7系统搜索dota2游戏协调服务器中究竟该怎么解决?其实不难根据下面的操作步骤就可以解决问题1:DOTA2服务器蹦了之后,进入DOTA2,发现最顶端先是提示:“搜索DOTA2协调服务器中…”…

  • 金士顿u盘修复工具怎么用_用diskgenius修复u盘

    金士顿u盘修复工具怎么用_用diskgenius修复u盘重新启动电脑后再插入读卡器试试左栏将列出电脑曾连接的所有网络名称对了,量产工具里有个CDROMSIZE,哪个是干嘛的?怎么填?u盘视频怎么在手机上播放不了怎么办到此就启动成功了,adatau盘量产,其实也不难,只是苹果电脑可能对某些启动盘有些挑剔,大家选择合适的准没问题。云顶攻略在WinPE系统的桌面上,鼠标右击我的电脑”—>管理”—>磁盘管理”,可以看到WinPE系笔记本u盘安装…

  • ng-repeat出现环路输出Duplicates in a repeater are not allowed. Use &#39;track by&#39; expression to specify unique

    ng-repeat出现环路输出Duplicates in a repeater are not allowed. Use &#39;track by&#39; expression to specify unique

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号