Соединитель искры Greenplum org. postgresql .util.PSQLException: ОШИБКА: ошибка при записи данных в gpfdist - PullRequest
0 голосов
/ 25 марта 2020

У меня кластер Greenplum на Azure, к которому я пытаюсь подключиться с помощью spark с моего локального компьютера (используя Pivotal Greenplum Spark Connector).

Я делаю что-то подобное в моем scala код:

var options = Map[String, String]()
options += ("url" -> url)
options += ("user" -> credential("user"))
options += ("password" -> credential("password"))
options += ("partitionColumn" -> partitionColumn)

sqlContext.read.format("greenplum").options(options).load()

Для целей тестирования я создал пользователя:

DROP USER IF EXISTS user1;
CREATE USER user1 CREATEEXTTABLE (type='writable') PASSWORD 'p@ss0rd';

Затем использовал этого пользователя для создания базы данных / таблицы следующим образом

drop table if exists sample;
create table public.sample (id serial, big bigint, wee smallint, stuff text) distributed by (id) ;
insert into sample (big) values (generate_series(1,100));
update sample set wee = 0; 
update sample set wee = 1 where mod(id,7)=1;
update sample set stuff = substr('abcdefghijklmnopqrstuvwxyz',1,mod(wee,13));

Теперь, когда я выполняю свой искровой код с учетными данными Greenplum, кажется (при работе в режиме отладки), что код успешно читает метаданные таблицы (он выбирает столбцы и типы), но чтение строк завершается неудачно с SQLSTATE(08006), ErrorCode(0) , Вот трассировка стека:

2020-03-24 19:04:31,168 WARN [Executor task launch worker for task 0] com.zaxxer.hikari.pool.ProxyConnection - HikariPool-1 - Connection org.postgresql.jdbc.PgConnection@14fab679 marked as broken because of SQLSTATE(08006), ErrorCode(0)
org.postgresql.util.PSQLException: ERROR: error when writing data to gpfdist http://127.0.0.1:60352/spark_e0aa1f0c8646f023_fffec8bf08e0054d_driver_261, quit after 8 tries  (seg1 172.21.0.4:6001 pid=25909)
    at org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2310)
    at org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2023)
    at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:217)
    at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:421)
    at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:318)
    at org.postgresql.jdbc.PgStatement.executeUpdate(PgStatement.java:294)
    at com.zaxxer.hikari.pool.ProxyStatement.executeUpdate(ProxyStatement.java:120)
    at com.zaxxer.hikari.pool.HikariProxyStatement.executeUpdate(HikariProxyStatement.java)
    at io.pivotal.greenplum.spark.jdbc.Jdbc$$anonfun$2.apply(Jdbc.scala:81)
    at io.pivotal.greenplum.spark.jdbc.Jdbc$$anonfun$2.apply(Jdbc.scala:79)
    at resource.AbstractManagedResource$$anonfun$5.apply(AbstractManagedResource.scala:88)
    at scala.util.control.Exception$Catch$$anonfun$either$1.apply(Exception.scala:125)
    at scala.util.control.Exception$Catch$$anonfun$either$1.apply(Exception.scala:125)
    at scala.util.control.Exception$Catch.apply(Exception.scala:103)
    at scala.util.control.Exception$Catch.either(Exception.scala:125)
    at resource.AbstractManagedResource.acquireFor(AbstractManagedResource.scala:88)
    at resource.ManagedResourceOperations$class.apply(ManagedResourceOperations.scala:26)
    at resource.AbstractManagedResource.apply(AbstractManagedResource.scala:50)
    at resource.DeferredExtractableManagedResource$$anonfun$tried$1.apply(AbstractManagedResource.scala:33)
    at scala.util.Try$.apply(Try.scala:192)
    at resource.DeferredExtractableManagedResource.tried(AbstractManagedResource.scala:33)
    at io.pivotal.greenplum.spark.jdbc.Jdbc$.copyTable(Jdbc.scala:83)
    at io.pivotal.greenplum.spark.externaltable.GreenplumRowIterator.liftedTree1$1(GreenplumRowIterator.scala:105)
    at io.pivotal.greenplum.spark.externaltable.GreenplumRowIterator.<init>(GreenplumRowIterator.scala:104)
    at io.pivotal.greenplum.spark.GreenplumRDD.compute(GreenplumRDD.scala:49)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
    at org.apache.spark.rdd.RDD$$anonfun$7.apply(RDD.scala:337)
    at org.apache.spark.rdd.RDD$$anonfun$7.apply(RDD.scala:335)
    at org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1165)
    at org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1156)
    at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:1091)
    at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1156)
    at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:882)
    at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:335)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:286)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
    at org.apache.spark.scheduler.Task.run(Task.scala:121)
    at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:402)
    at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:408)
    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)

Я пытался открыть порты в кластере Greenplum на Azure, но это не помогло enter image description here

Любые подсказки

Ответы [ 2 ]

1 голос
/ 25 марта 2020

Как отметил @denalex в своем ответе, запуск спарка с вашего локального компьютера может быть не самым простым, потому что Greenplum не будет знать, куда отправлять данные.

Другим подходом будет ускорение кластера Spark. или хост на Azure и сделать его видимым для Greenplum в той же сети.

1 голос
/ 25 марта 2020

Соединитель Spark запускает сервер gpfdist на каждом рабочем узле Spark, определяет адрес / имя хоста машины, на которой работает работник, и сообщает об этом в Greenplum, чтобы Greenplum мог отправлять на него данные. В вашем случае это разрешилось до 127.0.0.1, так как вы работаете на локальном компьютере, и Greenplum на Azure не удалось подключиться к серверу gpfdist на http://127.0.0.1: 60352

Для этого для работы ваши сотрудники Spark должны быть доступны с Azure через маршрутизируемый IP-адрес или разрешаемое имя хоста DNS. Вы можете указать, использовать ли имя хоста или IP-адрес (и какой сетевой интерфейс использовать для получения IP-адреса), указав параметры, описанные здесь: https://greenplum-spark.docs.pivotal.io/1-6/using_the_connector.html#addrcfg

Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.
...