Не удалось сохранить Ignite Spark в режиме YARN - PullRequest
Новая
       23

Не удалось сохранить Ignite Spark в режиме YARN

0 голосов
/ 17 мая 2019

Я использую Ignite 2.3, пытаюсь загрузить данные из файлов паркета hdfs в кэш Ignite с помощью Spark. Данные загружены, но не удалось сохранить их в Ignite Cache.

Если мы используем локальный режим вместо режима искровой ярды, он будет работать без проблем. Вопрос будет в том, почему в режиме YARN вычисления распределяются по кластерам, почему Ignite все еще нужен Ignite_Home? как это исправить? Какой-нибудь рабочий пример в режиме YARN?

Мой env: JDK 8, Ignite 2.3, Spark 2.1.0.cloudera1

моя искровая оболочка: 

env JAVA_HOME="/usr/java/jdk1.8.0_181-amd64" spark2-shell \
  --master yarn \
  --deploy-mode client \
  --driver-memory 12g \
  --conf spark.driver.extraJavaOptions="-DmavenTest=true" \
  --driver-class-path ~/libs/resources \
  --jars my.jar

Данные загружены в Spark, но не удалось сохранить их в Ignite Cache. 

val myDataRDD = ... //Spark RDD    
val igniteCluster = "10.xxx.xxx.xxx:64789"
val igniteContext = new IgniteContext(spark.sparkContext, () => new IgniteConfiguration()
      .setDiscoverySpi(new TcpDiscoverySpi().setIpFinder(new TcpDiscoveryVmIpFinder().setAddresses(List(igniteCluster).asJava)))
      .setWorkDirectory("/tmp/service")
      .setClientMode(true)
      .setDeploymentMode(DeploymentMode.CONTINUOUS)
      .setPeerClassLoadingEnabled(true)
)

val cfg: CacheConfiguration[MyDataKey, MyData] = new CacheConfiguration("cacheName");
cfg.setIndexedTypes(MyDataKey, MyData);
cfg.setCacheMode(PARTITIONED);
cfg.setAtomicityMode(ATOMIC);
cfg.setStatisticsEnabled(true);
cfg.setBackups(1);
cfg.setRebalanceMode(ASYNC);
cfg.setRebalanceBatchSize(2 * 1024 * 1024); //2MB
cfg.setRebalanceBatchesPrefetchCount(8);
val cacheRDD = igniteContext.fromCache[MyDataKey, MyData](cfg)
cacheRDD.savePairs(myDataRDD, true) //Failed

Сообщение об ошибке: 

Caused by: org.apache.ignite.IgniteException: Invalid Ignite installation home folder: /myproject/ROOT
  at org.apache.ignite.internal.util.IgniteUtils.convertException(IgniteUtils.java:966)
  at org.apache.ignite.Ignition.getOrStart(Ignition.java:417)
  at org.apache.ignite.spark.IgniteContext.ignite(IgniteContext.scala:143)
  at org.apache.ignite.spark.impl.IgniteAbstractRDD.ensureCache(IgniteAbstractRDD.scala:39)
  at org.apache.ignite.spark.IgniteRDD.compute(IgniteRDD.scala:59)
  at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
  at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
  at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
  at org.apache.spark.scheduler.Task.run(Task.scala:99)
  at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:322)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  at java.lang.Thread.run(Thread.java:748)
...