关于云计算:Flink的sink实战之三cassandra3

https://github.com/zq2599/blog_demos

内容：所有原创文章分类汇总及配套源码，波及 Java、Docker、Kubernetes、DevOPS 等；

本文是《Flink 的 sink 实战》系列的第三篇，次要内容是体验 Flink 官网的 cassandra connector，整个实战如下图所示，咱们先从 kafka 获取字符串，再执行 wordcount 操作，而后将后果同时打印和写入 cassandra：

《Flink 的 sink 实战之一：初探》
《Flink 的 sink 实战之二：kafka》
《Flink 的 sink 实战之三：cassandra3》
《Flink 的 sink 实战之四：自定义》

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

cassandra：3.11.6
kafka：2.4.0（scala：2.12）
jdk：1.8.0_191
flink：1.9.2
maven：3.6.0
flink 所在操作系统：CentOS Linux release 7.7.1908
cassandra 所在操作系统：CentOS Linux release 7.7.1908
IDEA：2018.3.5 (Ultimate Edition)

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

先创立 keyspace 和 table：

cqlsh 登录 cassandra：

cqlsh 192.168.133.168

创立 keyspace（3 正本）：

CREATE KEYSPACE IF NOT EXISTS example
    WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '3'};

建表：

CREATE TABLE IF NOT EXISTS example.wordcount (
    word text,
    count bigint,
    PRIMARY KEY(word)
    );

启动 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/blo…：

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

这个 git 我的项目中有多个文件夹，本章的利用在 flinksinkdemo 文件夹下，如下图红框所示：

flink 官网的 connector 反对两种形式写入 cassandra：

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

接下来别离应用这两种形式；

《Flink 的 sink 实战之二：kafka》中创立了 flinksinkdemo 工程，在此持续应用；
在 pom.xml 中减少 casandra 的 connector 依赖：

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

另外还要增加 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>

新增 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");
    }
}

上述代码中，从 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 写入，即业务逻辑中的数据结构实例被写入 cassandra，无需指定 SQL：

实现 POJO 写入数据库，须要 datastax 库的反对，在 pom.xml 中减少以下依赖：

<dependency>
  <groupId>com.datastax.cassandra</groupId>
  <artifactId>cassandra-driver-core</artifactId>
  <version>3.1.4</version>
  <classifier>shaded</classifier>
  <!-- Because the shaded JAR uses the original POM, you still need
                 to exclude this dependency explicitly: -->
  <exclusions>
    <exclusion>
    <groupId>io.netty</groupId>
    <artifactId>*</artifactId>
    </exclusion>
  </exclusions>
</dependency>

请留神下面配置的 exclusions 节点，依赖 datastax 的时候，依照官网领导对 netty 相干的间接依赖做排除，官网地址：https://docs.datastax.com/en/…
创立带有数据库相干注解的实体类 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);

        // 连贯 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<WordCount> result = dataStream
                .flatMap(new FlatMapFunction<String, WordCount>() {
                    @Override
                    public void flatMap(String s, Collector<WordCount> collector) throws Exception {String[] words = s.toLowerCase().split("\\s");

                        for (String word : words) {if (!word.isEmpty()) {
                                //cassandra 的表中，每个 word 都是主键，因而不能为空
                                collector.collect(new WordCount(word, 1L));
                            }
                        }
                    }
                })
                .keyBy("word")
                .timeWindow(Time.seconds(5))
                .reduce(new ReduceFunction<WordCount>() {
                    @Override
                    public WordCount reduce(WordCount wordCount, WordCount t1) throws Exception {return new WordCount(wordCount.getWord(), wordCount.getCount() + t1.getCount());
                    }
                });

        result.addSink(new PrintSinkFunction<>())
                .name("print Sink")
                .disableChaining();

        CassandraSink.addSink(result)
                .setHost("192.168.133.168")
                .setMapperOptions(() -> new Mapper.Option[] {Mapper.Option.saveNullFields(true) })
                .build()
                .name("cassandra Sink")
                .disableChaining();

        env.execute("kafka-2.4 source, cassandra-3.11.6 sink, pojo");
    }

}

从上述代码可见，和后面的 Tuple 写入类型有很大差异，为了筹备好 POJO 类型的数据集，除了 flatMap 的匿名类入参要改写，还要写好 reduce 办法的匿名类入参，并且还要调用 setMapperOptions 设置映射规定；
编译构建后，上传 jar 到 flink，并且指定工作类为 CassandraPojoSink：

清理之前的数据，在 cassandra 的 cqlsh 上执行 TRUNCATE example.wordcount;
像之前那样发送字符串音讯到 kafka：

查看数据库，发现后果合乎预期：

DAG 和 SubTask 状况如下：

至此，flink 的后果数据写入 cassandra 的实战就实现了，心愿能给您一些参考；

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

关于云计算:Flink的sink实战之三cassandra3

欢送拜访我的 GitHub

本篇概览

全系列链接

软件版本

对于 cassandra

筹备 cassandra 的 keyspace 和表

筹备 kafka 的 topic

源码下载

两种写入 cassandra 的形式

开发 (Tuple 写入)

开发 (POJO 写入)

欢送关注公众号：程序员欣宸

Just My Socks（注册教程内含优惠码）

关于云计算:Flink的sink实战之三cassandra3

欢送拜访我的 GitHub

本篇概览

全系列链接

软件版本

对于 cassandra

筹备 cassandra 的 keyspace 和表

筹备 kafka 的 topic

源码下载

两种写入 cassandra 的形式

开发 (Tuple 写入)

开发 (POJO 写入)

欢送关注公众号：程序员欣宸

Just My Socks（注册教程 内含优惠码）

Just My Socks（注册教程内含优惠码）