当前位置:网站首页>Flink的sink实战之三:cassandra3

Flink的sink实战之三:cassandra3

2020-11-10 17:30:43 程序猿欧文

欢迎访问我的GitHub

https://github.com/zq2599/blog_demos

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

本篇概览

本文是《Flink的sink实战》系列的第三篇,主要内容是体验Flink官方的cassandra connector,整个实战如下图所示,我们先从kafka获取字符串,再执行wordcount操作,然后将结果同时打印和写入cassandra:
在这里插入图片描述

全系列链接

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

软件版本

本次实战的软件版本信息如下:

  1. cassandra:3.11.6
  2. kafka:2.4.0(scala:2.12)
  3. jdk:1.8.0_191
  4. flink:1.9.2
  5. maven:3.6.0
  6. flink所在操作系统:CentOS Linux release 7.7.1908
  7. cassandra所在操作系统:CentOS Linux release 7.7.1908
  8. IDEA:2018.3.5 (Ultimate Edition)

关于cassandra

本次用到的cassandra是三台集群部署的集群,搭建方式请参考《ansible快速部署cassandra3集群》

准备cassandra的keyspace和表

先创建keyspace和table:

  1. cqlsh登录cassandra:
cqlsh 192.168.133.168
  1. 创建keyspace(3副本):
CREATE KEYSPACE IF NOT EXISTS example WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '3'};
  1. 建表:
CREATE TABLE IF NOT EXISTS example.wordcount ( word text, count bigint, PRIMARY KEY(word) );

准备kafka的topic

  1. 启动kafka服务;
  2. 创建名为test001的topic,参考命令如下:
./kafka-topics.sh \--create \--bootstrap-server 127.0.0.1:9092 \--replication-factor 1 \--partitions 1 \--topic test001
  1. 进入发送消息的会话模式,参考命令如下:
./kafka-console-producer.sh \--broker-list kafka:9092 \--topic test001
  1. 在会话模式下,输入任意字符串然后回车,都会将字符串消息发送到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:

  1. Tuple类型写入:将Tuple对象的字段对齐到指定的SQL的参数中;
  2. POJO类型写入:通过DataStax,将POJO对象对应到注解配置的表和字段中;

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

开发(Tuple写入)

  1. 《Flink的sink实战之二:kafka》中创建了flinksinkdemo工程,在此继续使用;
  2. 在pom.
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-cassandra_2.11</artifactId> <version>1.10.0</version></dependency>
  1. 另外还要添加flink-streaming-scala依赖,否则编译CassandraSink.addSink这段代码会失败:
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId> <version>${flink.version}</version> <scope>provided</scope></dependency>
  1. 新增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<String> flinkKafkaConsumer = new FlinkKafkaConsumer<>(    "test001",    new SimpleStringSchema(),    properties  );  //指定从最新位置开始消费,相当于放弃历史消息  flinkKafkaConsumer.setStartFromLatest();  //通过addSource方法得到DataSource  DataStream<String> dataStream = env.addSource(flinkKafkaConsumer);  DataStream<Tuple2<String, Long>> result = dataStream    .flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {        @Override        public void flatMap(String value, Collector<Tuple2<String, Long>> out) {         String[] words = value.toLowerCase().split("\\s");         for (String word : words) {          //cassandra的表中,每个word都是主键,因此不能为空          if (!word.isEmpty()) {           out.collect(new Tuple2<String, Long>(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"); }}
  1. 上述代码中,从kafka取得数据,做了word count处理后写入到cassandra,注意addSink方法后的一连串API(包含了数据库连接的参数),这是flink官方推荐的操作,另外为了在Flink web UI看清楚DAG情况,这里调用disableChaining方法取消了operator chain,生产环境中这一行可以去掉;
  2. 编码完成后,执行mvn clean package -U -DskipTests构建,在target目录得到文件flinksinkdemo-1.0-SNAPSHOT.jar;

版权声明
本文为[程序猿欧文]所创,转载请带上原文链接,感谢
https://my.oschina.net/mikeowen/blog/4711336