学习 Flink（十三）：Cassandra Connector

Flink 支持 Cassandra 作为 Sink。

依赖

编辑 pom.xml 文件，添加依赖：

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

Sink

Cassandra Sink 底层使用了 DataStax Java Driver。支持 CQL + Tuple 和 ORM 两种方式写入 Cassandra。

CQL + Tuple

已知 Tuple 有两个元素，第一个元素为 id，第二个元素为 name。

写入:chestnut:：

CassandraSink.addSink(stream)  
        .setHost("127.0.0.1")
        .setClusterBuilder(new ClusterBuilder() {
            @Override
            protected Cluster buildCluster(Cluster.Builder builder) {
                return builder.withCredentials("username", "password").build();
            }
        })
        .setQuery("INSERT INTO dm.user(id, name) values (?, ?);")
        .build();

ORM

定义 Java Bean：

@Table(keyspace = "dm", name = "user")
public class Test {

    @Column(name = "id")
    private Long id;

    @Column(name = "name")
    private String name;

    public Long getId() {
        return id;
    }

    public void setId(Long id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }
}

写入:chestnut:：

CassandraSink.addSink(stream)  
        .setHost("127.0.0.1")
        .setClusterBuilder(new ClusterBuilder() {
            @Override
            protected Cluster buildCluster(Cluster.Builder builder) {
                return builder.withCredentials("username", "password").build();
            }
        })
        .setMapperOptions(() -> new Mapper.Option[]{Mapper.Option.saveNullFields(true)})
        .build();

Q&A

本地运行，报错 java: cannot access org.apache.flink.streaming.api.scala.DataStream

编辑 pom.xml 文件，添加依赖：

<dependency>  
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-scala_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>

异常 java.lang.IllegalArgumentException: No support for the type of the given DataStream: GenericType

CassandraSink 输入类型必须为以下任意一种：

• Flink Java Tuple
• Scala case classe
• Row
• POJO

详情参考文档： Data Types & Serialization - Apache Flink Document 。

参考

• Apache Cassandra Connector - Apache Flink Document