а я говорил что надо более жёсткого бота поставить

Тут не оплачен

так есть бесплатный же

с капчей

Привет! А кто-то знает как конкретно df.repartition(10) шаффлит данные? Что конкретно используется для вычисления хеша и как его руками воспроизвести?

Думаешь там есть оптимизации прям такие? Я бы просто добавил rownum на каждой ноде и по остатку от деления на число нод отсылал бы на соответствующие ноды

Ага, ну нет, работает оно не так

partition = key.hashCode () % numPartitions

В случае датафрейма ключ это что?

ищу )

индексирование проекта спарка — боль

если мне память не изменяет, там рандом % новому числу партиций

Короче походу правда посередине.

Рандом должен быть нерандомный))
Для стабильного разделения сидируется номером партиции RDD, ограничен количеством партиций.

Формально там берётся хэшкод у объекта

  def getPartition(key: Any): Int = key match {
    case null => 0
    case _ => Utils.nonNegativeMod(key.hashCode, numPartitions)
  }

Но ключём выступает Row, видимо, у которого хэшкод не определён

А потом начинаем читать код и видим что всё сложнее, и, например, в UnsafeShuffleWriter ключём является значение из первой колонки

Но нас скорее всего интересует всё-таки

    def getPartitionKeyExtractor(): InternalRow => Any = newPartitioning match {
      case RoundRobinPartitioning(numPartitions) =>
        // Distributes elements evenly across output partitions, starting from a random partition.
        var position = new Random(TaskContext.get().partitionId()).nextInt(numPartitions)
        (row: InternalRow) => {
          // The HashPartitioner will handle the `mod` by the number of partitions
          position += 1
          position
        }
      case h: HashPartitioning =>
        val projection = UnsafeProjection.create(h.partitionIdExpression :: Nil, outputAttributes)
        row => projection(row).getInt(0)
      case RangePartitioning(sortingExpressions, _) =>
        val projection = UnsafeProjection.create(sortingExpressions.map(_.child), outputAttributes)
        row => projection(row)
      case SinglePartition => identity
      case _ => sys.error(s"Exchange not implemented for $newPartitioning")
    }

@ivanychev ↑

Если раундробин — то в точности как сказал Андрей