adding Cassandra as sink in Flink error : All host(s) tried for query failed

Question

I was following up with an example at https://ci.apache.org/projects/flink/flink-docs-release-1.4/dev/connectors/cassandra.html to connect Cassandra as sink in Flink

My code for is shown below

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

I am getting the following error

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)

Answer 1

Not sure if this is always required, but the way that I set up my CassandraSink is like this:

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

I have annotated POJOs that are returned by the dataStream so I don't need the query, but you would just include ".setQuery(...)" after the ".addSink(...)" line.

Answer 2

The exception simply indicates that the example program cannot reach the C* database.

1. flink-cassandra-connector offers streaming API to connect to designated C* database. Thus, you need to have a C* instance running.
2. Each streaming job is pushed/serialized to the node that Task Manager runs at. In your example, you assume C* is running on the same node as the TM node. An alternative is to change the C* address from 127.0.0.1 to a public address.