Как исправить Не удалось открыть собственное соединение с Cassandra по адресу {server ip}: 9042 - PullRequest
Новая
       27

Как исправить Не удалось открыть собственное соединение с Cassandra по адресу {server ip}: 9042

0 голосов
/ 29 марта 2019

Я пытаюсь соединить искру и Кассандру, используя спарк-кассандра-разъем. соединение устанавливается, но когда я пытаюсь выполнить операции на JavaRDD, я сталкиваюсь. 

   java.io.IOException: Failed to open native connection to Cassandra at {10.0.21.92}:9042

Вот конфигурация и код, который я пытаюсь реализовать: 

    SparkConf sparkConf = new SparkConf().setAppName("Data Transformation").set("spark.serializer","org.apache.spark.serializer.KryoSerializer").setMaster("local[4]");
    sparkConf.set("spark.cassandra.connection.host", server ip);
    sparkConf.set("spark.cassandra.connection.port", "9042");
    sparkConf.set("spark.cassandra.connection.timeout_ms", "5000");
    sparkConf.set("spark.cassandra.read.timeout_ms", "200000");
    sparkConf.set("spark.cassandra.auth.username", user_name);
    sparkConf.set("spark.cassandra.auth.password", password);

    JavaSparkContext sparkContext = new JavaSparkContext(sparkConf);

И ниже код, где я выполняю операцию на javardd: 

    CassandraJavaRDD<CassandraRow> cassandraRDD = CassandraJavaUtil.javaFunctions(sparkContext).cassandraTable(keySpaceName, tableName);
    JavaRDD<GenericTriggerEntity> rdd = cassandraRDD.map(new Function<CassandraRow, GenericTriggerEntity>() {

    private static final long serialVersionUID = -165799649937652815L; 

    @Override
    public GenericTriggerEntity call(CassandraRow row) throws Exception {
    GenericTriggerEntity genericTriggerEntity = new GenericTriggerEntity();
    if(row.getString("end") != null)                        genericTriggerEntity.setEnd(row.getString("end"));
    if(row.getString("key") != null)
    genericTriggerEntity.setKey(row.getString("key"));
    genericTriggerEntity.setKeyspacename(row.getString("keyspacename"));
    genericTriggerEntity.setPartitiondeleted(row.getString("partitiondeleted"));
    genericTriggerEntity.setRowdeleted(row.getString("rowDeleted"));
    genericTriggerEntity.setRows(row.getString("rows"));
    genericTriggerEntity.setStart(row.getString("start"));
    genericTriggerEntity.setTablename("tablename");
    genericTriggerEntity.setTriggerdate(row.getString("triggerdate"));
    genericTriggerEntity.setTriggertime(row.getString("triggertime"));
    genericTriggerEntity.setUuid(row.getUUID("uuid"));
    return genericTriggerEntity;
    }               

    });

Вот операция JavaRDD, которую я выполняю 


    JavaRDD<String> jsonDataRDDwords = rdd.flatMap(s -> Arrays.asList(SPACE.split((CharSequence) s)));
    JavaPairRDD<String, Integer> jsonDataRDDones = jsonDataRDDwords.mapToPair(s -> new Tuple2<>(s, 1));
    JavaPairRDD<String, Integer> jsonDataRDDcounts = jsonDataRDDones.reduceByKey((i1, i2) -> i1 + i2);
    List<Tuple2<String, Integer>> jsonDatRDDoutput = jsonDataRDDcounts.collect();

Я даже пытался telnet на сервер Cassandra, порт открыт.

Я могу установить соединение, но затем во время выполнения reduByKey получаю указанное выше исключение.

Я не могу понять, в чем проблема. Что-то не так в операции javardd. Любая помощь будет оценена. Заранее спасибо.

1 Ответ

0 голосов
/ 29 марта 2019

Вышеуказанная ошибка произошла из-за некоторой проблемы с зависимостью ядра привода cassandra. решил это, добавив метрическую зависимость в мой pom.xml 

<dependency>
<groupId>io.dropwizard.metrics</groupId>
<artifactId>metrics-core</artifactId>
<version>3.2.2</version>
</dependency>
...