1

我在https://ci.apache.org/projects/flink/flink-docs-release-1.4/dev/connectors/cassandra.html跟进了一个示例，将 Cassandra 连接为 Flink 中的接收器

我的代码如下所示

public class writeToCassandra {

    private static final String CREATE_KEYSPACE_QUERY = "CREATE KEYSPACE test WITH replication= {'class':'SimpleStrategy', 'replication_factor':1};";
    private static final String createTable = "CREATE TABLE test.cassandraData(id varchar, heart_rate varchar, PRIMARY KEY(id));" ;


    private final static Collection<String> collection = new ArrayList<>(50);

    static {
        for (int i = 1; i <= 50; ++i) {
            collection.add("element " + i);
        }
    }

    public static void main(String[] args) throws Exception {


        //setting the env variable to local
        StreamExecutionEnvironment envrionment = StreamExecutionEnvironment.createLocalEnvironment(1);


        DataStream<Tuple2<String, String>> dataStream = envrionment
                .fromCollection(collection)
                .map(new MapFunction<String, Tuple2<String, String>>() {

                    final String mapped = " mapped ";
                    String[] splitted;

                    @Override
                    public Tuple2<String, String> map(String s) throws Exception {
                        splitted = s.split("\\s+");
                        return Tuple2.of(
                                UUID.randomUUID().toString(),
                                splitted[0] + mapped + splitted[1]
                        );
                    }
                });


        CassandraSink.addSink(dataStream)
                .setQuery("INSERT INTO test.cassandraData(id,heart_rate) values (?,?);")
                .setHost("127.0.0.1")
                .build();


        envrionment.execute();

    } //main




} //writeToCassandra

我收到以下错误

Caused by: com.datastax.driver.core.exceptions.NoHostAvailableException: All host(s) tried for query failed (tried: /127.0.0.1:9042 (com.datastax.driver.core.exceptions.TransportException: [/127.0.0.1] Cannot connect))
    at com.datastax.driver.core.ControlConnection.reconnectInternal(ControlConnection.java:231)
4

2 回答 2

1

不确定这是否总是需要的，但我设置 CassandraSink 的方式是这样的：

CassandraSink
    .addSink(dataStream)
    .setClusterBuilder(new ClusterBuilder() {
        @Override
        protected Cluster buildCluster(Cluster.Builder builder) {
            return Cluster.builder()
                .addContactPoints(myListOfCassandraUrlsString.split(","))
                .withPort(portNumber)
                .build();
        }
    })
    .build();

我已经注释了 dataStream 返回的 POJO，所以我不需要查询，但您只需在“.addSink(...)”行之后包含“.setQuery(...)”。

于 2017-09-01T16:19:47.470 回答
0

该异常仅表明示例程序无法访问 C* 数据库。

  1. flink-cassandra-connector 提供流式 API 来连接指定的 C* 数据库。因此，您需要运行一个 C* 实例。
  2. 每个流式作业都被推送/序列化到任务管理器运行的节点。在您的示例中，您假设 C* 与 TM 节点在同一节点上运行。另一种方法是将 C* 地址从 127.0.0.1 更改为公共地址。
于 2017-10-02T13:25:54.950 回答