Spark 2.0.0: чтение с Кассандры в режиме кластера - PullRequest
Новая
       24

Spark 2.0.0: чтение с Кассандры в режиме кластера

0 голосов
/ 27 сентября 2018

У меня проблемы с запуском Spark-приложения, которое читает данные из Cassandra в Spark 2.0.0.

Мой код работает следующим образом: 

DataFrameReader readerCassandra = SparkContextUtil.getInstance().read() 
                    .format("org.apache.spark.sql.cassandra")
                    .option("spark.cassandra.connection.host", [DATABASE_IP])
                    .option("spark.cassandra.connection.port", [DATABASE_PORT]);

final Map<String,String> map = new HashMap<String,String>();

map.put("table", "MyTable");
map.put("keyspace", "MyKeyspace");

public final  StructType schema = DataTypes.createStructType(
        new StructField[] { DataTypes.createStructField("id", DataTypes.StringType, true),
            DataTypes.createStructField("timestamp", DataTypes.TimestampType, true),
            DataTypes.createStructField("value", DataTypes.DoubleType, true)
        });

final Dataset<Row> dataset = readerCassandra.schema(schema).options(map).load(); 
dataset.show(false);

Я хочу запустить этот код вкластер.Мой кластер использует spark-2.0.2-bin-hadoop2.7 (нет версии spark-2.0.0 на http://spark.apache.org/downloads.html).

Сначала я отправляю его в режиме клиента с помощью следующего сценария: 

#!/bin/bash

sparkMaster=local[*]
mainClass=package.MainClass

jar=/path/to/myJar-with-dependencies.jar

driverPort=7079
blockPort=7082

deployMode=client

$SPARK_HOME/bin/spark-submit \
  --conf "spark.driver.port=${driverPort}"\
  --conf "spark.blockManager.port=${blockPort}"\
  --class $mainClass \
  --master $sparkMaster \
  --deploy-mode $deployMode \
  --jars /path/to/jars/spark-cassandra-connector_2.11-2.0.0.jar \
  $jar

Когда я это делаю, все работает хорошо. Но теперь я хочу запустить свое приложение в режиме кластера.

Поэтому я немного модифицирую свой скрипт отправки, установив sparkMaster с моим главным IPи deployMode в «кластер».

Когда я отправляю свою заявку, в журналах драйверов почти мгновенно появляется следующая ошибка: 

Exception in thread "main" java.lang.reflect.InvocationTargetException
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at org.apache.spark.deploy.worker.DriverWrapper$.main(DriverWrapper.scala:58)
        at org.apache.spark.deploy.worker.DriverWrapper.main(DriverWrapper.scala)
Caused by: java.lang.ClassNotFoundException: Failed to find data source: org.apache.spark.sql.cassandra. Please find packages at https://cwiki.apache.org/confluence/display/SPARK/Third+Party+Projects
        at org.apache.spark.sql.execution.datasources.DataSource.lookupDataSource(DataSource.scala:148)
        ...

Caused by: java.lang.ClassNotFoundException: org.apache.spark.sql.cassandra.DefaultSource
        at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
        ...

Примечание:

  • У меня все еще есть ошибка с кластером только одного Рабочего на той же машине, что и мой Мастер.
  • Сначала я использовал Spark 2.3.1, и у меня не было проблем с запуском моего кода вРежим кластера (с использованием spark-cassandra-connector_2.11-2.3.1.jar в --jars).
  • Я пробовал несколько банок в --jars, например: spark-cassandra-connector_2.11-2.0.0.jar, spark-cassandra-connector_2.11-2.0.2.jar, spark-cassandra-connector_2.11-2.3.1.jar, spark-cassandra-connector-java_2.11-1.5.1.jar, но ни один из них не работал.
  • Некоторые другие банки задаются в параметре --jars и учитываются

1 Ответ

0 голосов
/ 27 сентября 2018

Вместо этого вам может потребоваться указать путь как file:///path/to/jars/spark-cassandra-connector_2.11-2.0.0.jar - в этом случае он будет разослан исполнителям через HTTP-сервер драйвера.В противном случае он ожидает, что файл уже скопирован вами на все машины, чтобы избежать копирования самим процессом.Подробности смотрите в документации Spark ...

Я бы порекомендовал просто создать uberjar со всеми зависимостями (кроме Spark) и отправить его - в таких случаях было бы меньше боли.

...