学习 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。

写入🌰:

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;
    }
}

写入🌰:

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

参考