Я создал RDD бизнес-объявлений по городским кодам, разделенным по французским департаментам, таким образом:
JavaRDD<Etablissement> etablissements = sc.textFile(source.getAbsolutePath())
.map(this.mapCsvVersEtablissements)
.filter(etablissement -> etablissement.hasAnomalie(SeveriteAnomalie.ERREUR, false) == false
&& etablissement.isAnnulationLogique() == false)
.mapToPair(etablissement ->
new Tuple2<>(etablissement.getAdresses().get(0).getCodeCommune(), etablissement))
.partitionBy(this.partitionParCodeCommune)
.persist(StorageLevel.MEMORY_ONLY());
Затем для каждого из моих городов я делаю lookup(codeCommune) для этогоСДР, чтобы найти все имеющиеся у них бизнес-документы и выполнить над ними операции.
Когда СДР впервые присоединяется, он создается, и использование памяти на моем компьютере составляет тогда 10 GB (среди этого использования памяти встречается 5 GB из моего Windows сеанса, который работает).
Во время обработки одного из этих городов у меня есть эти журналы, как пример для одного города:
Début du recensement des activités de la commune Brazey-en-Morvan (21102).
1/2 debut lookup(21102)
Collecte des établissements de la commune 21102...
Starting job: lookup at AcquisitionEntrepriseService.java:385
Got job 6749 (lookup at AcquisitionEntrepriseService.java:385) with 1 output partitions
Final stage: ResultStage 13499 (lookup at AcquisitionEntrepriseService.java:385)
Parents of final stage: List(ShuffleMapStage 13498)
Missing parents: List()
Submitting ResultStage 13499 (ShuffledRDD[6] at partitionBy at AcquisitionEntrepriseService.java:265), which has no missing parents
Block broadcast_6751 stored as values in memory (estimated size 3.3 KB, free 6.2 GB)
Block broadcast_6751_piece0 stored as bytes in memory (estimated size 2.0 KB, free 6.2 GB)
Added broadcast_6751_piece0 in memory on SEVEN-PC:63687 (size: 2.0 KB, free: 6.2 GB)
Created broadcast 6751 from broadcast at DAGScheduler.scala:1161
Submitting 1 missing tasks from ResultStage 13499 (ShuffledRDD[6] at partitionBy at AcquisitionEntrepriseService.java:265) (first 15 tasks are for partitions Vector(21))
Adding task set 13499.0 with 1 tasks
Starting task 0.0 in stage 13499.0 (TID 6888, localhost, executor driver, partition 21, PROCESS_LOCAL, 7662 bytes)
Running task 0.0 in stage 13499.0 (TID 6888)
Found block rdd_6_21 locally
Finished task 0.0 in stage 13499.0 (TID 6888). 20805 bytes result sent to driver
Finished task 0.0 in stage 13499.0 (TID 6888) in 21 ms on localhost (executor driver) (1/1)
Removed TaskSet 13499.0, whose tasks have all completed, from pool
ResultStage 13499 (lookup at AcquisitionEntrepriseService.java:385) finished in 0,023 s
Job 6749 finished: lookup at AcquisitionEntrepriseService.java:385, took 0,023287 s
Fin de la collecte de tous les établissements de la commune 21102.
2/2 @26ms fin de l'examen des 35 établissements, début de l'écriture en base.
@50ms Fin du recensement des activités de la commune Brazey-en-Morvan (21102).
1) Я счастлив, потому что мой раздел работает хорошо.Города из департамента 21 (Côte-d'Or) взяты из хорошего раздела:
Found block rdd_6_21 locally
2) 6751, показанный в качестве номера работы, является рангом города среди 36 000 Iнадо справиться, все нормально.Кажется, что Spark создает два этапа каждый раз, когда этапы достигают рангов 13498 и 13499.
Однако я не понимаю:
a) Почему я читаю это?
Submitting ResultStage 13499 (ShuffledRDD[6] at partitionBy at AcquisitionEntrepriseService.java:265), which has no missing parents
partitionBy уже выполнено при создании RDD и уже должно быть в кеше?
b) Что эти трансляции хранятся в памяти?
c) Увеличение использования ЦП и памяти при поиске.
После примерно 10 000 поисков у меня возникает ошибка нехватки памяти, GC limit exceed: Использование памяти увеличилось на 4,4 GB для 24 GB доступно (вся память, используемая компьютером, на 14,4 GB тогда) и ЦП на 99% при сбое программы.
Я видел, что некоторые очистки автоматически появлялись нажурналы консоли.Но нужно ли мне выполнять некоторые ручные чистки самостоятельно во время моей пакетной работы, говоря Spark, чтобы так или иначе удалить предыдущие lookup результаты, которые, возможно, были сохранены где-то в памяти, если это то, что происходит?
С уважением,