Является ли детектор Spark RDD c для набора элементов в каждом разделе?

Question

Я не могу найти много документации по обеспечению порядка секционирования - я просто хочу убедиться, что с учетом набора детерминированных c преобразований (выходные строки всегда одинаковы), разделы всегда получают одинаковый набор элементов, если базовый набор данных не меняется Это возможно?

Не нужно сортировать: пример будет после того, как набор преобразований будет применен к СДР, теперь это выглядит так -> (A, B, C, D, E, F, G)

И если бы мой spark.default.parallelism был 2 или 3, набор элементов всегда был бы либо: (A, B, C, D), (E, F, G ) или (A, B), (C, D), (E, F, G) соответственно.

Это связано с тем, что мои исполнители будут вызывать некоторые побочные эффекты в зависимости от раздела / множество элементов, над которыми он работает, и я хочу убедиться, что приложение Spark идемпотентно. (тот же побочный эффект, если он перезапускается)

Редактировать: По-видимому, перераспределение DF является детерминированным c, а раздел RDD - нет (Spark 2.4.4).

def f1(rdds):
    rows = list(rdds)
    stats_summary = [{
        'origin': str(row['origin']),
        'dest': str(row['dest']),
        'start_time': analysis_date.isoformat(),
        'value': row['count']
    } for row in rows]

    stats_summary.sort(key=lambda t: (t['start_time'], t['origin'], t['dest']))

    rtn = "partition size: {}, first: ({}, {}), last: ({}, {})".format(
        len(rows), 
        stats_summary[0]["origin"], stats_summary[0]["dest"],
        stats_summary[-1]["origin"], stats_summary[-1]["dest"])
    return [rtn]

repartition_rdd_res = unq_statistics.rdd \
                                    .repartition(10) \
                                    .mapPartitions(f1) \
                                    .collect()

repartition_df_res = unq_statistics.repartition(10) \
                                   .rdd \
                                   .mapPartitions(f1) \
                                   .collect()

repartition_rdd_res4 = ['partition size: 131200, first: (-1, -1), last: (999, -1)',
 'partition size: 131209, first: (-1, 1014), last: (996, 996)',
 'partition size: 131216, first: (-1, 1021), last: (999, 667)',
 'partition size: 131218, first: (-1, 1008), last: (991, 1240)',
 'partition size: 131222, first: (-1, 1001), last: (994, 992)',
 'partition size: 131229, first: (-1, 1007), last: (994, 890)',
 'partition size: 131233, first: (-1, 1004), last: (991, -1)',
 'partition size: 131235, first: (-1, 1005), last: (999, 1197)',
 'partition size: 131237, first: (-1, 100), last: (999, 997)',
 'partition size: 131240, first: (-1, 1010), last: (994, -1)']

repartition_rdd_res3 = ['partition size: 131200, first: (-1, -1), last: (999, -1)',
 'partition size: 131209, first: (-1, 1006), last: (994, 2048)',
 'partition size: 131216, first: (-1, 1002), last: (996, 996)',
 'partition size: 131218, first: (-1, 1017), last: (999, 667)',
 'partition size: 131222, first: (-1, 1008), last: (994, 890)',
 'partition size: 131229, first: (-1, 1000), last: (99, 96)',
 'partition size: 131233, first: (-1, 1001), last: (994, 992)',
 'partition size: 131235, first: (-1, 1009), last: (990, 1601)',
 'partition size: 131237, first: (-1, 1004), last: (994, -1)',
 'partition size: 131240, first: (-1, 1003), last: (999, 997)']

repartition_rdd_res2 = ['partition size: 131200, first: (-1, 1013), last: (991, 2248)',
 'partition size: 131209, first: (-1, 1007), last: (999, 667)',
 'partition size: 131216, first: (-1, 100), last: (99, 963)',
 'partition size: 131218, first: (-1, 1002), last: (999, 997)',
 'partition size: 131222, first: (-1, 101), last: (996, 996)',
 'partition size: 131229, first: (-1, -1), last: (991, 1240)',
 'partition size: 131233, first: (-1, 1006), last: (999, 1197)',
 'partition size: 131235, first: (-1, 1001), last: (994, 992)',
 'partition size: 131237, first: (-1, 1019), last: (999, -1)',
 'partition size: 131240, first: (-1, 1017), last: (991, -1)']

repartition_df_res2 = ['partition size: 131222, first: (-1, 1023), last: (996, 996)',
 'partition size: 131223, first: (-1, 1003), last: (999, 667)',
 'partition size: 131223, first: (-1, 1012), last: (990, 990)',
 'partition size: 131224, first: (-1, -1), last: (999, 1558)',
 'partition size: 131224, first: (-1, 100), last: (99, 98)',
 'partition size: 131224, first: (-1, 1008), last: (99, 968)',
 'partition size: 131224, first: (-1, 1018), last: (999, 997)',
 'partition size: 131225, first: (-1, 1006), last: (994, 992)',
 'partition size: 131225, first: (-1, 101), last: (990, 935)',
 'partition size: 131225, first: (-1, 1013), last: (999, 1197)']

Answer 1

Давайте посмотрим на источник и, в частности, его часть в случайном порядке:

...
if (shuffle) {
  /** Distributes elements evenly across output partitions, starting from a random partition. */
  val distributePartition = (index: Int, items: Iterator[T]) => {
    var position = new Random(hashing.byteswap32(index)).nextInt(numPartitions)
    items.map { t =>
      // Note that the hash code of the key will just be the key itself. The HashPartitioner
      // will mod it with the number of total partitions.
      position = position + 1
      (position, t)
    }
  } : Iterator[(Int, T)]
  ...

Как видно распределение элементов из данного исходного раздела N в X Целевые разделы - это простое приращение (позднее по модулю X), начиная с некоторого числа, которое зависит только от этого N и, следовательно, предопределено. Поэтому, если ваш СДР-источник не изменился, результат repartition(X) также должен быть одинаковым каждый раз.

Answer 2

В дополнение к тому, что все говорят:

• Датафрейм основан на СДР.
• Вы говорите: перераспределение кадров данных является детерминированным c.
• Если СДР перераспределение не будет детерминированным c, тогда перераспределение Dataframe также не будет.

Answer 3

Внутренне Spark использует секционер по умолчанию ( HashPartitioner в зависимости от данных) для разделения данных, который использует ha sh для определения того, к какому разделу принадлежит данный элемент. Таким образом, вы можете сказать, что элемент данных всегда будет go для одного и того же раздела, учитывая, что количество разделов одинаково, потому что, если количество разделов изменяется, это также повлияет на ha sh.

Answer 4

Распределение записей по разделам не обязательно должно быть равномерным. Количество разделов гарантировано, и количество записей в каждом разделе будет примерно одинаковым. Это не имеет значения для любой операции. Если по какой-либо причине произойдет случайное перемешивание, новые разделы будут воссозданы.

Скажем (A, B, C, D, E, F, G) разделены на 2 как (A, B, * 1005). *, D) и (E, F, G). Если обработка, выполняемая исполнителем (E, F, G), прекратится, Spark перезапустит ее и попытается повторно обработать (E, F, G). Если этот исполнитель не подлежит восстановлению, тогда все задание не будет выполнено, и оно начнется заново с разбиения (A, B, C, D, E, F, G) на 2 части и перезапускает обработку. Во 2-й попытке он может быть разделен на (A, B, C) и (D, E, F, G). Конечный результат обработки будет таким же.

Answer 5

Относительно отказоустойчивости, если раздел терпит неудачу, он будет повторно передан другому исполнителю, поэтому разделы не упорядочены.

В Spark есть некоторый внутренний механизм для расчета наилучшего плана выполнения с учетом имеющихся у вас данных. Таким образом, вы не можете предсказать порядок и содержание разделов.