Apache Cassandra Connector(连接器)

这个Connectors提供sink即写数据到一个Cassandra数据库中
为了使用这个Connectors，将下面的依赖添加到你的工程中:

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-cassandra_2.10</artifactId>
  <version>1.3.0</version>
</dependency>

注意:streaming connectors目前还不是二进制发布包的一部分，请参考此处来了解如何在分布式执行中关联到这些connectors。

安装Cassandra

请参考这篇文档

Cassandra Sink

Flink的Cassandra sink通过这个静态的方法CassandraSink.addSink(DataStream input)来创建,这个方法会返回一个CassandraSinkBuilder，它提供了方法来更深入地配置sink。
下面的配置方法可以被使用:
　　1、setQuery(String query)
　　2、setHost(String host[, int port])
　　3、setClusterBuilder(ClusterBuilder builder)
　　4、enableWriteAheadLog([CheckpointCommitter committer])
　　5、build()
setQuery()方法设置了为sink接收到的每个值执行的query语句，setHost()设置要去连接的Cassandra的host/port，此方法用于简单的用例，setclusterbuilder()设置了用来配置连接到Cassandra的cluster builder，setHost()的功能可以被这个方法替代。enableWriteAheadLog()是个可选的方法，为非确定性算法提供精确处理(exactly-once)。

例如:
Java 代码:

CassandraSink.addSink(input)
  .setQuery("INSERT INTO example.values (id, counter) values (?, ?);")
  .setClusterBuilder(new ClusterBuilder() {
    @Override
    public Cluster buildCluster(Cluster.Builder builder) {
      return builder.addContactPoint("127.0.0.1").build();
    }
  })
  .build();

Scala代码

CassandraSink.addSink(input)
  .setQuery("INSERT INTO example.values (id, counter) values (?, ?);")
  .setClusterBuilder(new ClusterBuilder() {
    override def buildCluster(builder: Cluster.Builder): Cluster = {
      builder.addContactPoint("127.0.0.1").build()
    }
  })
  .build()

Cassandra sink支持使用DataStax注释的Tuple和POJO，Flink会自动去探测输入数据的类型:
一个使用DataStax注释的Pojo例子:

@Table(keyspace= "test", name = "mappersink")
public class Pojo implements Serializable {

  private static final long serialVersionUID = 1038054554690916991L;

  @Column(name = "id")
  private long id;
  @Column(name = "value")
  private String value;

  public Pojo(long id, String value){
    this.id = id;
    this.value = value;
  }

  public long getId() {
    return id;
  }

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

  public String getValue() {
    return value;
  }

  public void setValue(String value) {
    this.value = value;
  }
}