Ошибка соединения Spark 2.2 с огромным набором данных

В настоящее время у меня возникают проблемы при попытке объединить (внутренний) огромный набор данных (654 ГБ) с меньшим (535 МБ) с помощью Spark DataFrame API.

Я транслирую меньший набор данных рабочим узлам с помощью функции broadcast ().

Я не могу объединить эти два набора данных. Вот образец ошибок, которые я получил:

19/04/26 19:39:07 INFO executor.CoarseGrainedExecutorBackend: Got assigned task 1315
19/04/26 19:39:07 INFO executor.Executor: Running task 25.1 in stage 13.0 (TID 1315)
19/04/26 19:39:07 INFO output.FileOutputCommitter: File Output Committer Algorithm version is 1
19/04/26 19:39:07 INFO datasources.SQLHadoopMapReduceCommitProtocol: Using output committer class org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter
19/04/26 19:39:07 INFO datasources.FileScanRDD: Reading File path: SOMEFILEPATH, range: 3087007744-3221225472, partition values: [empty row]
19/04/26 19:39:17 INFO datasources.FileScanRDD: Reading File path: SOMEFILEPATH, range: 15971909632-16106127360, partition values: [empty row]
19/04/26 19:39:24 WARN hdfs.DFSClient: DFSOutputStream ResponseProcessor exception  for block isi_hdfs_pool:blk_4549851005_134218728
java.io.IOException: Connection reset by peer
    at sun.nio.ch.FileDispatcherImpl.read0(Native Method)
    at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39)
    at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223)
    at sun.nio.ch.IOUtil.read(IOUtil.java:197)
    at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:380)
    at org.apache.hadoop.net.SocketInputStream$Reader.performIO(SocketInputStream.java:57)
    at org.apache.hadoop.net.SocketIOWithTimeout.doIO(SocketIOWithTimeout.java:142)
    at org.apache.hadoop.net.SocketInputStream.read(SocketInputStream.java:161)
    at org.apache.hadoop.net.SocketInputStream.read(SocketInputStream.java:131)
    at org.apache.hadoop.net.SocketInputStream.read(SocketInputStream.java:118)
    at java.io.FilterInputStream.read(FilterInputStream.java:83)
    at java.io.FilterInputStream.read(FilterInputStream.java:83)
    at org.apache.hadoop.hdfs.protocolPB.PBHelper.vintPrefixed(PBHelper.java:2280)
    at org.apache.hadoop.hdfs.protocol.datatransfer.PipelineAck.readFields(PipelineAck.java:244)
    at org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer$ResponseProcessor.run(DFSOutputStream.java:733)
19/04/26 19:39:27 ERROR util.Utils: Aborting task
com.univocity.parsers.common.TextWritingException: Error writing row.
Internal state when error was thrown: recordCount=458089, recordData=["SOMEDATA"]
    at com.univocity.parsers.common.AbstractWriter.throwExceptionAndClose(AbstractWriter.java:916)
    at com.univocity.parsers.common.AbstractWriter.writeRow(AbstractWriter.java:706)
    at org.apache.spark.sql.execution.datasources.csv.UnivocityGenerator.write(UnivocityGenerator.scala:82)
    at org.apache.spark.sql.execution.datasources.csv.CsvOutputWriter.write(CSVFileFormat.scala:139)
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$SingleDirectoryWriteTask.execute(FileFormatWriter.scala:327)
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$3.apply(FileFormatWriter.scala:258)
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$3.apply(FileFormatWriter.scala:256)
    at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1375)
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala:261)
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1$$anonfun$apply$mcV$sp$1.apply(FileFormatWriter.scala:191)
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1$$anonfun$apply$mcV$sp$1.apply(FileFormatWriter.scala:190)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
    at org.apache.spark.scheduler.Task.run(Task.scala:108)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335)
    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)
Caused by: java.lang.IllegalStateException: Error closing the output.
    at com.univocity.parsers.common.AbstractWriter.close(AbstractWriter.java:861)
    at com.univocity.parsers.common.AbstractWriter.throwExceptionAndClose(AbstractWriter.java:903)
    at com.univocity.parsers.common.AbstractWriter.writeRow(AbstractWriter.java:811)
    at com.univocity.parsers.common.AbstractWriter.writeRow(AbstractWriter.java:704)
    ... 15 more
Caused by: java.io.IOException: All datanodes DatanodeInfoWithStorage[10.241.209.34:585,null,DISK] are bad. Aborting...
    at org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.setupPipelineForAppendOrRecovery(DFSOutputStream.java:1109)
    at org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.processDatanodeError(DFSOutputStream.java:871)
    at org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.run(DFSOutputStream.java:401)
19/04/26 19:39:27 WARN util.Utils: Suppressing exception in catch: Failed on local exception: java.io.IOException: Connection reset by peer; Host Details : local host is: "SOMENODEHOST"; destination host is: "SOMEDESTINATIONHOST":SOMEPORT; 
java.io.IOException: Failed on local exception: java.io.IOException: Connection reset by peer; Host Details : local host is: "SOMENODEHOST"; destination host is: "SOMEDESTINATIONHOST":SOMEPORT; 
    at org.apache.hadoop.net.NetUtils.wrapException(NetUtils.java:776)
    at org.apache.hadoop.ipc.Client.call(Client.java:1479)
    at org.apache.hadoop.ipc.Client.call(Client.java:1412)
    at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:229)
    at com.sun.proxy.$Proxy17.delete(Unknown Source)
    at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.delete(ClientNamenodeProtocolTranslatorPB.java:540)
    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.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:191)
    at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:102)
    at com.sun.proxy.$Proxy18.delete(Unknown Source)
    at org.apache.hadoop.hdfs.DFSClient.delete(DFSClient.java:2044)
    at org.apache.hadoop.hdfs.DistributedFileSystem$14.doCall(DistributedFileSystem.java:707)
    at org.apache.hadoop.hdfs.DistributedFileSystem$14.doCall(DistributedFileSystem.java:703)
    at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81)
    at org.apache.hadoop.hdfs.DistributedFileSystem.delete(DistributedFileSystem.java:714)
    at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.abortTask(FileOutputCommitter.java:568)
    at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.abortTask(FileOutputCommitter.java:557)
    at org.apache.spark.internal.io.HadoopMapReduceCommitProtocol.abortTask(HadoopMapReduceCommitProtocol.scala:159)
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$1.apply$mcV$sp(FileFormatWriter.scala:266)
    at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1384)
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala:261)
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1$$anonfun$apply$mcV$sp$1.apply(FileFormatWriter.scala:191)
    at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1$$anonfun$apply$mcV$sp$1.apply(FileFormatWriter.scala:190)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
    at org.apache.spark.scheduler.Task.run(Task.scala:108)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335)
    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)
Caused by: java.io.IOException: Connection reset by peer
    at sun.nio.ch.FileDispatcherImpl.read0(Native Method)
    at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39)
    at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223)
    at sun.nio.ch.IOUtil.read(IOUtil.java:197)
    at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:380)
    at org.apache.hadoop.net.SocketInputStream$Reader.performIO(SocketInputStream.java:57)
    at org.apache.hadoop.net.SocketIOWithTimeout.doIO(SocketIOWithTimeout.java:142)
    at org.apache.hadoop.net.SocketInputStream.read(SocketInputStream.java:161)
    at org.apache.hadoop.net.SocketInputStream.read(SocketInputStream.java:131)
    at java.io.FilterInputStream.read(FilterInputStream.java:133)
    at java.io.FilterInputStream.read(FilterInputStream.java:133)
    at org.apache.hadoop.ipc.Client$Connection$PingInputStream.read(Client.java:520)
    at java.io.BufferedInputStream.fill(BufferedInputStream.java:246)
    at java.io.BufferedInputStream.read(BufferedInputStream.java:265)
    at java.io.DataInputStream.readInt(DataInputStream.java:387)
    at org.apache.hadoop.ipc.Client$Connection.receiveRpcResponse(Client.java:1084)
    at org.apache.hadoop.ipc.Client$Connection.run(Client.java:979)

Перед тем, как объединить большой набор данных с меньшим, я попытался объединить 10 000 записей первой со всей меньшей (535 МБ). У меня была ошибка «Истекло время ожидания фьючерса [300 с]».

Затем я увеличил значение переменной spark.sql.broadcastTimeout до 3600 с. Все работало нормально. Но когда я пытаюсь объединить его со всем набором данных (654 ГБ), появляется сообщение об ошибке, которое вы видите наверху (исключение TextWriting).

Мои вопросы:

  • Как я могу более эффективно контролировать свои искровые работы? И как мне действовать?

  • Как вы думаете, что вызывает эту ошибку? Как я могу это решить?

Ниже вы найдете некоторую информацию о кластере, выполнении и конфигурации искрового задания.

Некоторая информация / контекст:

Я работаю в производственной среде (см. Конфигурацию кластера ниже). Я не могу обновить свою версию Spark. У меня нет пользовательского интерфейса Spark или Yarn для отслеживания моих рабочих мест. Все, что я могу достать, - это журналы пряжи.

Образец кода

def readCsv(spark: SparkSession, path: String): DataFrame = {
    spark.read
      .option("header", true)
      .option("escape", "\"")
      .option("mode", "FAILFAST")
      .csv(path)
  }

    val uh_months = readCsv(spark, input_dir_terro + "HDFS_PATH_OF_ALL_THE_CSV_FILES")
      .withColumnRenamed("NUM", "NO_NUM")
      .where(col("BEWC").isin(
        LIST OF VALUES))
      .withColumn("january", lit("1960-01-01"))

    val uh = uh_months
      .withColumn("UHDIN", datediff(to_date(unix_timestamp(col("UHDIN_YYYYMMDD"), "yyyyMMdd").cast(TimestampType)),
        to_date(unix_timestamp(col("january"), "yyyy-MM-dd").cast(TimestampType))))
      //      .withColumn("DVA_1", to_date((unix_timestamp(col("DVA"), "ddMMMyyyy")).cast(TimestampType)))
      .withColumn("DVA_1", date_format(col("DVA"), "dd/MM/yyyy"))
      .drop("UHDIN_YYYYMMDD")
      .drop("january")
      .drop("DVA")

    val uh_joined = uh.join(broadcast(smallDF), "KEY")
      .select(
        uh.col("*"),
        smallDF.col("PSP"),
        smallDF.col("minrel"),
        smallDF.col("Label"),
        smallDF.col("StartDate"))
      .where(smallDF.col("PSP").isNotNull)
      .withColumnRenamed("DVA_1", "DVA")
      .where(col("BKA").isNotNull)

smallDF - это фрейм данных размером 535 МБ, полученный после некоторых агрегатов и преобразований.

План выполнения

    == Physical Plan ==
*Project [NO_NUM#252, DEV#153, DEBCRED#154, BDGRORI#155, BDGREUR#156, BEWC#157, MSG30_NL#158, SCAPMV#159, USERID#160, MMED#161, TNUM#162, NMTGP#163, BKA#164, CATEXT#165, SEQETAT#166, ACCTYPE#167, BRAND#168, FAMILY#169, SUBFAMILY#170, FORCED_DVA#172, BYBANK#173, CPTE_PROTEGE#174, HOURMV#175, RDFB#176, ... 30 more fields]
+- *BroadcastHashJoin [NO_NUM#252], [NO_NUM#13], Inner, BuildRight
   :- *Project [NUM#152 AS NO_NUM#252, DEV#153, DEBCRED#154, BDGRORI#155, BDGREUR#156, BEWC#157, MSG30_NL#158, SCAPMV#159, USERID#160, MMED#161, TNUM#162, NMTGP#163, BKA#164, CATEXT#165, SEQETAT#166, ACCTYPE#167, BRAND#168, FAMILY#169, SUBFAMILY#170, FORCED_DVA#172, BYBANK#173, CPTE_PROTEGE#174, HOURMV#175, RDFB#176, ... 26 more fields]
   :  +- *Filter ((BEWC#157 INSET (25003,25302,25114,20113,12017,20108,25046,12018,15379,15358,11011,20114,10118,12003,25097,20106,20133,10133,10142,15402,25026,25345,28023,15376,25019,28004,21701,25001,11008,15310,15003,28020,22048,15470,25300,25514,25381,25339,15099,25301,28005,28026,25098,25018,15323,25376,15804,15414,25344,25102,15458,15313,28002,25385,22051,25214,15031,12005,15425,20145,22011,15304,25027,14020,11007,25901,15343,22049,20112,12031,20127,15339,25421,15432,28025,25340,25325,20150,28011,25368,25304,22501,25369,28022,15098,12032,15375,25002,25008,10116,10101,22502,25090,15004,20105,12030,22503,15095,22007,15809,15342,15311,25216,10103,20122,11019,20142,15097,20147,20149,25005,25205,25380,15380,10120,25015,15384,11003,10110,25016,15090,25307,15001,25390,15312,10115,25219,15806,15459,12016,15359,15395,15302,12021,11701,10111,10148,25379,15807,10102,25352,25355,12010,25095,25394,20101,25413,15385,25322,28027,11026,15533,25201,25371,10128,11028,12020,15819,10143,28028,10123,10125,11020,25029,10122,25343,15015,12033,25014,12012,25024,25375,11023,25501,25402,22001,15317,12014,16114,20501,15046,12001,12022,10104,10117,12002,25499,10145,10153,12011,15350,15300,10119,25305,15345,25374,11027,25430,28021,25202,10121,28024,25101,28001,15321,11025,25358,15333,15501,25533,15372,12008,11015,10114,10113,10112,15303,15320,28006,22002,25359,10132,15497,25353,11029,25425,15374,12019,25437,11022,15357,20148,20111,26114,25099,25354,10124,25303,11010,20120,20135,15820,15331,28029) && isnotnull(BKA#164)) && isnotnull(NUM#152))
   :     +- *FileScan csv [UHDIN_YYYYMMDD#151,NUM#152,DEV#153,DEBCRED#154,BDGRORI#155,BDGREUR#156,BEWC#157,MSG30_NL#158,SCAPMV#159,USERID#160,MMED#161,TNUM#162,NMTGP#163,BKA#164,CATEXT#165,SEQETAT#166,ACCTYPE#167,BRAND#168,FAMILY#169,SUBFAMILY#170,DVA#171,FORCED_DVA#172,BYBANK#173,CPTE_PROTEGE#174,... 26 more fields] Batched: false, Format: CSV, Location: InMemoryFileIndex[hdfs://SOMEHOST:SOMEPORT/SOMEPATH..., PartitionFilters: [], PushedFilters: [In(BEWC, [25003,25302,25114,20113,12017,20108,25046,12018,15379,15358,11011,20114,10118,12003,25..., ReadSchema: struct<UHDIN_YYYYMMDD:string,NUM:string,DEV:string,DEBCRED:string,BDGRORI:string,BDGREUR:string,B...
   +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]))
      +- *Project [NO_NUM#13, minrel#370, PSP#82, Label#105, StartDate#106]
         +- *SortMergeJoin [PSP#381], [PSP#82], Inner
            :- *Sort [PSP#381 ASC NULLS FIRST], false, 0
            :  +- Exchange hashpartitioning(PSP#381, 200)
            :     +- *Project [PSP#381, NO_NUM#13, minrel#370]
            :        +- SortMergeJoin [PSP#381, C_SNUM#14, minrel#370, NO_NUM#13], [NO_PSP#47, C_SNUM_1#387, C_NRELPR#50, NO_NUM_1#400], LeftOuter
            :           :- *Sort [PSP#381 ASC NULLS FIRST, C_SNUM#14 ASC NULLS FIRST, minrel#370 ASC NULLS FIRST, NO_NUM#13 ASC NULLS FIRST], false, 0
            :           :  +- Exchange hashpartitioning(PSP#381, C_SNUM#14, minrel#370, NO_NUM#13, 200)
            :           :     +- SortAggregate(key=[NO_PSP#12, C_SNUM#14, NO_NUM#13], functions=[min(C_NRELPR#15)])
            :           :        +- *Sort [NO_PSP#12 ASC NULLS FIRST, C_SNUM#14 ASC NULLS FIRST, NO_NUM#13 ASC NULLS FIRST], false, 0
            :           :           +- Exchange hashpartitioning(NO_PSP#12, C_SNUM#14, NO_NUM#13, 200)
            :           :              +- SortAggregate(key=[NO_PSP#12, C_SNUM#14, NO_NUM#13], functions=[partial_min(C_NRELPR#15)])
            :           :                 +- *Sort [NO_PSP#12 ASC NULLS FIRST, C_SNUM#14 ASC NULLS FIRST, NO_NUM#13 ASC NULLS FIRST], false, 0
            :           :                    +- *Project [NO_PSP#12, C_SNUM#14, NO_NUM#13, C_NRELPR#15]
            :           :                       +- *Filter (((C_NRELPR#15 IN (001,006) && C_SNUM#14 IN (030,033)) && isnotnull(NO_PSP#12)) && isnotnull(NO_NUM#13))
            :           :                          +- *FileScan csv [NO_PSP#12,NO_NUM#13,C_SNUM#14,c_nrelpr#15] Batched: false, Format: CSV, Location: InMemoryFileIndex[hdfs://SOMEHOST:SOMEPORT/SOMEPATH..., PartitionFilters: [], PushedFilters: [In(c_nrelpr, [001,006]), In(C_SNUM, [030,033]), IsNotNull(NO_PSP), IsNotNull(NO_NUM)], ReadSchema: struct<NO_PSP:string,NO_NUM:string,C_SNUM:string,c_nrelpr:string>
            :           +- *Sort [NO_PSP#47 ASC NULLS FIRST, C_SNUM_1#387 ASC NULLS FIRST, C_NRELPR#50 ASC NULLS FIRST, NO_NUM_1#400 ASC NULLS FIRST], false, 0
            :              +- Exchange hashpartitioning(NO_PSP#47, C_SNUM_1#387, C_NRELPR#50, NO_NUM_1#400, 200)
            :                 +- *Project [NO_PSP#47, NO_NUM#48 AS NO_NUM_1#400, C_SNUM#49 AS C_SNUM_1#387, c_nrelpr#50]
            :                    +- *FileScan csv [NO_PSP#47,NO_NUM#48,C_SNUM#49,c_nrelpr#50] Batched: false, Format: CSV, Location: InMemoryFileIndex[hdfs://SOMEHOST:SOMEPORT/SOMEPATH..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<NO_PSP:string,NO_NUM:string,C_SNUM:string,c_nrelpr:string>
            +- *Sort [PSP#82 ASC NULLS FIRST], false, 0
               +- Exchange hashpartitioning(PSP#82, 200)
                  +- *Project [PSP#82, Label#105, StartDate#106]
                     +- *Filter isnotnull(PSP#82)
                        +- *FileScan csv [PSP#82,Label#105,StartDate#106] Batched: false, Format: CSV, Location: InMemoryFileIndex[hdfs://SOMEHOST:SOMEPORT/SOMEPATH..., PartitionFilters: [], PushedFilters: [IsNotNull(PSP)], ReadSchema: struct<PSP:string,Label:string,StartDate:string>

Версия Spark: 2.2

Конфигурация кластера:

  • 21 вычислительный узел (рабочий)
  • 8 ядер каждое
  • 64 ГБ ОЗУ на узел

Текущая конфигурация Spark:

-мастер: пряжа

-исполнитель-память: 42 ГБ

-исполнителей-ядер: 5

-драйвера память: 42 г

-количество исполнителей: 28

-spark.sql.broadcastTimeout = 3600

-spark.kryoserializer.buffer.max = 512

-spark.yarn.executor.memoryOverhead = 2400

-spark.driver.maxResultSize = 500 м

-spark.memory.storageFraction = 0,3

-spark.memory.fraction = 0,9

-spark.hadoop.fs.permissions.umask-mode = 007

Как выполняется работа:

Мы создаем артефакт (банку) с помощью IntelliJ, а затем отправляем его на сервер. Затем выполняется сценарий bash. Этот сценарий:

  • экспортировать некоторые переменные среды (SPARK_HOME, HADOOP_CONF_DIR, PATH и SPARK_LOCAL_DIRS)

  • запустите команду spark-submit со всеми параметрами, определенными в конфигурации spark выше

  • извлекает журналы пряжи приложения


apache-spark hdfs apache-spark-sql join cluster-computing
person Ali    schedule 30.04.2019    source источник
comment
Обновите свой вопрос, используя текущий код или образец, план выполнения (вы можете распечатать его с помощью df.explain ()) и, если возможно, DAG из Spark UI.   -  person abiratsis    schedule 02.05.2019
comment
Спасибо за ответ. Я добавил образец кода. Однако, как я уже упоминал, у меня нет доступа к Spark UI. Что касается плана выполнения, я сейчас его делаю и отредактирую вопрос, как только он у меня появится.   -  person Ali    schedule 02.05.2019
comment
Я добавил план выполнения   -  person Ali    schedule 02.05.2019
comment
Хорошо, спасибо. Ваша трансляция работает, это хорошие новости   -  person abiratsis    schedule 02.05.2019
comment
Как вы смогли определить, что трансляция работает нормально? Если бы этого не произошло, что бы я увидел или не увидел вместо этого?   -  person Ali    schedule 03.05.2019

Ответы (1)


arrow_upward
3
arrow_downward

Вот некоторые улучшения в вашем коде:

  1. Добавьте repartition на основе столбца KEY, к которому вы присоединяетесь с uh, количество разделов должно быть примерно 650GB / 500MB ~ 1300.
  2. Примените фильтрацию к своим наборам данных перед присоединением к ним, в вашем случае просто выполните предложения where перед оператором соединения.
  3. Необязательно cache небольшой набор данных
  4. Убедитесь, что небольшой набор данных будет транслироваться, т.е. вы можете попытаться сохранить и проверить его размер. Затем соответствующим образом отрегулируйте значение spark.broadcast.blockSize, возможно, увеличив его.

Вот как должен выглядеть ваш код с изменениями:

    val uh_months = readCsv(spark, input_dir_terro + "HDFS_PATH_OF_ALL_THE_CSV_FILES") 
      .withColumnRenamed("OLD_KEY", "KEY")
      .where(col("code").isin(LIST OF VALUES))
      .withColumn("january", lit("1960-01-01"))

    val uh = uh_months
      .withColumn("UHDIN", datediff(to_date(unix_timestamp(col("UHDIN_YYYYMMDD"), "yyyyMMdd").cast(TimestampType)),
        to_date(unix_timestamp(col("january"), "yyyy-MM-dd").cast(TimestampType))))
      //      .withColumn("field_1", to_date((unix_timestamp(col("field"), "ddMMMyyyy")).cast(TimestampType)))
      .withColumn("field_1", date_format(col("field"), "dd/MM/yyyy"))
      .drop("UHDIN_YYYYMMDD")
      .drop("january")
      .drop("field")
      .repartition(1300, $"KEY") //change 1: repartition based on KEY with 1300 (650GB/500MB~1300)

    //change 2: always prune as much information as possible before joining!
    val smallerDF = smallDF
                      .where(smallDF.col("ID").isNotNull && col("field_6").isNotNull)
                      .select("KEY", "ID", "field_3", "field_4", "field_5")

     //change 3: you can optionally cache the small dataset
    smallerDF.cache()

    //change 4: adjust spark.broadcast.blockSize i.e spark.conf.set("spark.broadcast.blockSize","16m"

    val uh_joined = uh.join(broadcast(smallerDF), "KEY")
      .select(
        uh.col("*"),
        smallerDF.col("ID"),
        smallerDF.col("field_3"),
        smallerDF.col("field_4"),
        smallerDF.col("field_5"))
      .withColumnRenamed("field_1", "field")

Последнее замечание, касающееся конфигурации вашего кластера. Я бы попытался увеличить num-executors как минимум на 32, поскольку уровень распараллеливания в таком большом кластере должен быть выше.

person abiratsis    schedule 02.05.2019
comment
Спасибо, действительно сработало! Я даже не установил параметр spark.broadcast.blockSize. Не могли бы вы объяснить несколько вещей: - зачем мне кэшировать небольшой набор данных, если я уже транслирую его на рабочие узлы? - вы разделили размер большого набора данных на размер меньшего. Как вы смогли определить, что это будет достаточное количество разделов для выполнения моего искрового задания? Это обычная практика, которую мне нужно делать, когда я объединяю два набора данных на основе ключа? - person Ali; 03.05.2019
comment
Привет, @Ali, 1. Вам не обязательно кэшировать небольшой набор данных, поэтому я называю его необязательным. Причина предположить, что, возможно, ваша программа использовала его позже 2. Я не разделил размер большого набора данных на маленький, я просто разделил размер большого на размер идеального размера раздела, который составляет 200 -500 МБ. Проверьте это ссылку для получения дополнительных сведений о размере разделов и о том, как указать правильный номер раздела. - person abiratsis; 03.05.2019
comment
3. Да, это способ присоединения к большим наборам данных на основе определенного ключа или комбинации ключей (второй вариант несколько сложнее). Посмотрите эту отличную статью от двух активных Spark участники :) - person abiratsis; 03.05.2019
comment
4. Вы хорошо поработали, вы не изменили spark.broadcast.blockSize, поскольку кажется, что небольшой набор данных уже соответствует размеру блока boadcast. Это было бы полезно, только если размер набора данных был больше, чем spark.broadcast.blockSize. - person abiratsis; 03.05.2019
comment
Большое спасибо @Alexandros за помощь и за статьи, это действительно полезно. - person Ali; 03.05.2019
comment
Я рад, что смог помочь @Ali - person abiratsis; 04.05.2019
comment
Я разместил еще один вопрос с проблемой сортировки, с которой я столкнулся в том же коде, если вы хотите взглянуть: Проблема со ссылкой - person Ali; 07.05.2019
comment
Привет, @Ali :) Я проверю и дам тебе знать - person abiratsis; 08.05.2019