Flink的sink实战之三：cassandra3(1)

准备kafka的topic

启动kafka服务；
创建名为test001的topic，参考命令如下：

./kafka-topics.sh \

–create \

–bootstrap-server 127.0.0.1:9092 \

–replication-factor 1 \

–partitions 1 \

–topic test001

进入发送消息的会话模式，参考命令如下：

./kafka-console-producer.sh \

–broker-list kafka:9092 \

–topic test001

在会话模式下，输入任意字符串然后回车，都会将字符串消息发送到broker；

源码下载

如果您不想写代码，整个系列的源码可在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文件夹下，如下图红框所示：

两种写入cassandra的方式

flink官方的connector支持两种方式写入cassandra：

Tuple类型写入：将Tuple对象的字段对齐到指定的SQL的参数中；
POJO类型写入：通过DataStax，将POJO对象对应到注解配置的表和字段中；

接下来分别使用这两种方式；

开发(Tuple写入)

《Flink的sink实战之二：kafka》中创建了flinksinkdemo工程，在此继续使用；
在pom.xml中增加casandra的connector依赖：

org.apache.flink

flink-connector-cassandra_2.11

1.10.0

另外还要添加flink-streaming-scala依赖，否则编译CassandraSink.addSink这段代码会失败：

org.apache.flink

flink-streaming-scala_${scala.binary.version}

${flink.version}

provided

新增CassandraTuple2Sink.java，这就是Job类，里面从kafka获取字符串消息，然后转成Tuple2类型的数据集写入cassandra，写入的关键点是Tuple内容和指定SQL中的参数的匹配：

package com.bolingcavalry.addsink;

import org.apache.flink.api.common.functions.FlatMapFunction;

import org.apache.flink.api.common.serialization.SimpleStringSchema;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.streaming.api.datastream.DataStream;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.streaming.connectors.cassandra.CassandraSink;

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import org.apache.flink.util.Collector;

import java.util.Properties;

public class CassandraTuple2Sink {

public static void main(String[] args) throws Exception {

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

//设置并行度

env.setParallelism(1);

//连接kafka用到的属性对象

Properties properties = new Properties();

//broker地址

properties.setProperty(“bootstrap.servers”, “192.168.50.43:9092”);

//zookeeper地址

properties.setProperty(“zookeeper.connect”, “192.168.50.43:2181”);

//消费者的groupId

properties.setProperty(“group.id”, “flink-connector”);

//实例化Consumer类

FlinkKafkaConsumer flinkKafkaConsumer = new FlinkKafkaConsumer<>(

“test001”,

new SimpleStringSchema(),

properties

);

//指定从最新位置开始消费，相当于放弃历史消息

flinkKafkaConsumer.setStartFromLatest();

//通过addSource方法得到DataSource

DataStream dataStream = env.addSource(flinkKafkaConsumer);

DataStream> result = dataStream

.flatMap(new FlatMapFunction>() {

@Override

public void flatMap(String value, Collector> out) {

String[] words = value.toLowerCase().split(“\s”);

for (String word : words) {

//cassandra的表中，每个word都是主键，因此不能为空

if (!word.isEmpty()) {

out.collect(new Tuple2(word, 1L));

}

)

.keyBy(0)

.timeWindow(Time.seconds(5))

.sum(1);

result.addSink(new PrintSinkFunction<>())

.name(“print Sink”)

.disableChaining();

CassandraSink.addSink(result)

.setQuery(“INSERT INTO example.wordcount(word, count) values (?, ?);”)

.setHost(“192.168.133.168”)

.build()

.name(“cassandra Sink”)

.disableChaining();

env.execute(“kafka-2.4 source, cassandra-3.11.6 sink, tuple2”);

}

上述代码中，从kafka取得数据，做了word count处理后写入到cassandra，注意addSink方法后的一连串API(包含了数据库连接的参数)，这是flink官方推荐的操作，另外为了在Flink web UI看清楚DAG情况，这里调用disableChaining方法取消了operator chain，生产环境中这一行可以去掉；
编码完成后，执行mvn clean package -U -DskipTests构建，在target目录得到文件flinksinkdemo-1.0-SNAPSHOT.jar；
在Flink的web UI上传flinksinkdemo-1.0-SNAPSHOT.jar，并指定执行类，如下图红框所示：

启动任务后DAG如下：

去前面创建的发送kafka消息的会话模式窗口，发送一个字符串"aaa bbb ccc aaa aaa aaa"；
查看cassandra数据，发现已经新增了三条记录，内容符合预期：

查看TaskManager控制台输出，里面有Tuple2数据集的打印结果，和cassandra的一致：

DAG上所有SubTask的记录数也符合预期：

开发(POJO写入)

接下来尝试POJO写入，即业务逻辑中的数据结构实例被写入cassandra，无需指定SQL：

实现POJO写入数据库，需要datastax库的支持，在pom.xml中增加以下依赖：

com.datastax.cassandra

cassandra-driver-core

3.1.4

shaded

io.netty

请注意上面配置的exclusions节点，依赖datastax的时候，按照官方指导对netty相关的间接依赖做排除，官方地址：https://docs.datastax.com/en/developer/java-driver/3.1/manual/shaded_jar/
创建带有数据库相关注解的实体类WordCount：

package com.bolingcavalry.addsink;

import com.datastax.driver.mapping.annotations.Column;

import com.datastax.driver.mapping.annotations.Table;

@Table(keyspace = “example”, name = “wordcount”)

public class WordCount {

@Column(name = “word”)

private String word = “”;

@Column(name = “count”)

private long count = 0;

public WordCount() {

}

public WordCount(String word, long count) {

this.setWord(word);

this.setCount(count);

}

public String getWord() {

return word;

}

public void setWord(String word) {

this.word = word;

}

public long getCount() {

return count;

}

public void setCount(long count) {

this.count = count;

}

@Override

public String toString() {

return getWord() + " : " + getCount();

}

然后创建任务类CassandraPojoSink：

package com.bolingcavalry.addsink;

import com.datastax.driver.mapping.Mapper;

import com.datastax.shaded.netty.util.Recycler;

import org.apache.flink.api.common.functions.FlatMapFunction;

import org.apache.flink.api.common.functions.ReduceFunction;

import org.apache.flink.api.common.serialization.SimpleStringSchema;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.streaming.api.datastream.DataStream;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.streaming.connectors.cassandra.CassandraSink;

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import org.apache.flink.util.Collector;

import java.util.Properties;

public class CassandraPojoSink {

public static void main(String[] args) throws Exception {

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

//设置并行度

env.setParallelism(1);

最后

自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。

深知大多数Java工程师，想要提升技能，往往是自己摸索成长，自己不成体系的自学效果低效漫长且无助。

因此收集整理了一份《2024年Java开发全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友，同时减轻大家的负担。

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上Java开发知识点，不论你是刚入门Android开发的新手，还是希望在技术上不断提升的资深开发者，这些资料都将为你打开新的学习之门！

如果你觉得这些内容对你有帮助，需要这份全套学习资料的朋友可以戳我获取！！

由于文件比较大，这里只是将部分目录截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且会持续更新！

进入阿里一直到现在。**

深知大多数Java工程师，想要提升技能，往往是自己摸索成长，自己不成体系的自学效果低效漫长且无助。

[外链图片转存中…(img-dZYhcPCC-1715307557114)]

[外链图片转存中…(img-S9vLCYxe-1715307557115)]

[外链图片转存中…(img-fQpExbsu-1715307557115)]

如果你觉得这些内容对你有帮助，需要这份全套学习资料的朋友可以戳我获取！！

由于文件比较大，这里只是将部分目录截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且会持续更新！

分类:热门推荐日期:2024-06-23浏览:1评论:0

准备kafka的topic

源码下载

两种写入cassandra的方式

开发(Tuple写入)

开发(POJO写入)

最后

在这里 地址

相关推荐