這個Connectors提供sink即寫數(shù)據到一個Cassandra數(shù)據庫中
為了使用這個Connectors棺聊，將下面的依賴添加到你的工程中:

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

注意:streaming connectors目前還不是二進制發(fā)布包的一部分，請參考此處來了解如何在分布式執(zhí)行中關聯(lián)到這些connectors栗柒。

安裝Cassandra

請參考這篇文檔

Cassandra Sink

Flink的Cassandra sink通過這個靜態(tài)的方法CassandraSink.addSink(DataStream input)來創(chuàng)建,這個方法會返回一個CassandraSinkBuilder，它提供了方法來更深入地配置sink誊抛。
下面的配置方法可以被使用:
　　1孵班、setQuery(String query)
　　2、setHost(String host[, int port])
　　3竖独、setClusterBuilder(ClusterBuilder builder)
　　4、enableWriteAheadLog([CheckpointCommitter committer])
　　5挤牛、build()
setQuery()方法設置了為sink接收到的每個值執(zhí)行的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，F(xiàn)link會自動去探測輸入數(shù)據的類型:
一個使用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;
  }
}