невозможно применить протокол gpfdist при работе с искрой

Question

Я пытаюсь прочитать данные из greenplum в HDFS, используя спарк. Для этого я использую файл jar: greenplum-spark_2.11-1.6.0.jar

Применение spark.read как показано ниже:

val yearDF = spark.read.format("io.pivotal.greenplum.spark.GreenplumRelationProvider").option("url", "jdbc:postgresql://1.2.3.166:5432/finance?ssl=true&sslfactory=org.postgresql.ssl.NonValidatingFactory").option("server.port","8020").option("dbtable", "tablename").option("dbschema","schema").option("user", "123415").option("password", "etl_123").option("partitionColumn","je_id").option("partitions",3).load().where("period_year=2017 and period_num=12 and source_system_name='SSS'").select(splitSeq map col:_*).withColumn("flagCol", lit(0))
yearDF.write.format("csv").save("hdfs://dev/apps/hive/warehouse/header_test_data/")

Когда я запускаю приведенный выше код, я получаю исключение:

Exception in thread "qtp1438055710-505" java.lang.OutOfMemoryError: GC overhead limit exceeded
19/03/05 12:29:08 WARN QueuedThreadPool:
java.lang.OutOfMemoryError: GC overhead limit exceeded
19/03/05 12:29:08 WARN QueuedThreadPool: Unexpected thread death: org.eclipse.jetty.util.thread.QueuedThreadPool$3@16273740 in qtp1438055710{STARTED,8<=103<=200,i=19,q=0}
19/03/05 12:36:03 ERROR Executor: Exception in task 0.0 in stage 4.0 (TID 8)
org.postgresql.util.PSQLException: ERROR: error when writing data to gpfdist http://1.2.3.8:8020/spark_6ca7d983d07129f2_db5510e67a8a6f78_driver_370, quit after 2 tries (url_curl.c:584)  (seg7 ip-1-3-3-196.ec2.internal:40003 pid=4062) (cdbdisp.c:1322)
    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)

Я применил шаги так же, как они были упомянуты в этой официальной документации

Ранее я использовал банку: greenplum.jar, которая работала нормально, но работала медленнее, когда передавала данные через GP Master. Jar: greenplum-spark_2.11-1.6.0.jar - это jar соединителя, который использует протокол gpfdist для передачи данных в HDFS.

Также IP-адреса изменяются в сообщении об исключении. Вы можете увидеть IP 1.2.3.166:5432 стать 1.2.3.8:8020, а также seg7 ip-1-3-3-196.ec2.internal:40003 pid=4062

При том же количестве исполнителей и памяти исполнителей я мог бы получать данные, используя greenplum.jar. Но сохранил все то же самое и просто изменил банку на greenplum-spark_2.11-1.6.0.jar только для того, чтобы увидеть это исключение Я пытался решить проблему, но я не понимаю этого явления вообще. Может кто-нибудь сообщить мне, как я могу решить эту проблему?

Answer 1

Можете ли вы увеличить количество разделов? В зависимости от размера таблицы вам может понадобиться увеличить количество разделов. Можете ли вы попробовать увеличить количество разделов до 30 и посмотреть, не возникли ли у вас проблемы с нехваткой памяти?

val yearDF = spark.read.format("io.pivotal.greenplum.spark.GreenplumRelationProvider").option("url", "jdbc:postgresql://1.2.3.166:5432/finance?ssl=true&sslfactory=org.postgresql.ssl.NonValidatingFactory").option("server.port","8020").option("dbtable", "tablename").option("dbschema","schema").option("user", "123415").option("password", "etl_123").option("partitionColumn","je_id").option("partitions",30).load().where("period_year=2017 and period_num=12 and source_system_name='SSS'").select(splitSeq map col:_*).withColumn("flagCol", lit(0))
yearDF.write.format("csv").save("hdfs://dev/apps/hive/warehouse/header_test_data/")

Answer 2

Разъем Greenplum-Spark предназначен для распараллеливания передачи данных между сегментами Greenplum и работниками Spark. Чтобы в полной мере использовать параллельную передачу данных, вам необходимо обеспечить достаточное количество памяти и рабочих искров, чтобы ускорить передачу данных. В противном случае вы можете использовать greenplum.jar, который использует один JDBC-коннектор для загрузки данных из HDFS в базу данных Greenplum через одного мастера Greenplum. Это значительно медленнее, когда вы используете загрузку данных в один мастер Greenplum.

Некоторые соображения: - Зависит от количества сегментов Greenplum, достаточно ли у вас рабочих / исполнителей Spark для получения или отправки данных между кластерами Spark и Greenplum?
- Зависит от памяти, выделенной рабочим / исполнителям Spark. Относится к документу " Tuning Spark "

Из журнала ошибок с этим сообщением «java.lang.OutOfMemoryError: Превышен предел накладных расходов GC», я могу предположить, что у вашего искрового работника / исполнителей недостаточно памяти. Вам все еще нужно настроить ваших сотрудников Spark, чтобы они могли распараллеливать загрузку данных из HDFS.