Я новичок в Apache Spark, и я знаю, что основная структура данных - это RDD. Теперь я пишу некоторые приложения, для которых требуется информация о позициях элемента. Например, после преобразования ArrayList в RDD (Java) для каждого целого в RDD мне нужно знать его (глобальный) индекс массива. Можно ли это сделать?
Как я знаю, для RDD существует функция take (int), поэтому я считаю, что позиционная информация сохраняется в RDD.
По существу, метод RDD zipWithIndex(), похоже, делает это, но он не сохранит первоначальный порядок данных, из которых был создан RDD. По крайней мере, вы получите стабильный заказ.
val orig: RDD[String] = ...
val indexed: RDD[(String, Long)] = orig.zipWithIndex()
Причина, по которой вы вряд ли найдете что-то, что сохраняет порядок в исходных данных, зарывается в API doc для zipWithIndex():
"Заменяет этот RDD своими индексами элементов. Сначала упорядочение по индексу раздела, а затем упорядочивание элементов в каждом раздел. Таким образом, первый элемент в первом разделе получает индекс 0 и последний элемент последнего раздела получает наибольший индекс. Эта похож на Scala zipWithIndex, но использует Long вместо Int как тип индекса. Этот метод должен запускать искровое задание, когда этот RDD содержит более одного раздела.
Итак, похоже, что исходный порядок отбрасывается. Если для вас важно сохранить исходный порядок, похоже, вам нужно добавить индекс перед созданием RDD.
В большинстве случаев я верю, что zipWithIndex() выполнит трюк и сохранит заказ. Прочтите комментарии еще раз. Я понимаю, что это точно означает сохранить порядок в RDD.
scala> val r1 = sc.parallelize(List("a", "b", "c", "d", "e", "f", "g"), 3)
scala> val r2 = r1.zipWithIndex
scala> r2.foreach(println)
(c,2)
(d,3)
(e,4)
(f,5)
(g,6)
(a,0)
(b,1)
В приведенном выше примере подтвердите это. Красный имеет 3 раздела, а с индексом 0, b с индексом 1 и т.д.