Как разбить Spark DataFrame на массивы numpy, чтобы использовать их в качестве входных данных сверточной нейронной сети тензорного потока?

Question

У меня есть искровый DataFrame «A», который выглядит так:

   customer         A         B         C
0         1  0.999471  0.746963  0.098438
1         1  0.986164  0.762772  0.715350
2         1  0.656797  0.975615  0.002074
3         2  0.892825  0.542348  0.648907
4         2  0.806137  0.806392  0.365054
5         2  0.707369  0.535159  0.205938
6         3  0.072516  0.381937  0.076068
7         3  0.442076  0.634749  0.918520
8         3  0.628006  0.842492  0.971928

Я хочу разделить фрейм данных на 3 высоты, 3 ширины и 1 глубину numpy массивы (похожие на изображения) по заказчику чтобы использовать их в качестве входных данных для сверточной нейронной сети тензорного потока. Например, для customer 1 желаемый np.array с shape (3, 3, 1) будет выглядеть так:

array([[[0.99947138],
        [0.74696311],
        [0.09843827]],

       [[0.98616396],
        [0.7627723 ],
        [0.71534965]],

       [[0.65679665],
        [0.97561509],
        [0.00207389]]])

Пока у меня есть следующий код (который, я думаю, неэффективно, так как на создание каждого изображения для того объема данных, который я обрабатываю, уходит почти 10 секунд)

customers = [1,2,3]
images = []
for customer in customers:
    img = A.filter(col("customer") == customer).select("A", "B", "C").toPandas().values.reshape(3,3,1)
    images.append(img)

Есть ли лучший способ сделать это?

Примечание: После этого я конвертирую список изображений в массив numpy, чтобы ввести его в сеть CNN.

Answer 1

Какие библиотеки вы используете для CNN? Если вы используете стандартные библиотеки, такие как keras или pytorch, то единственный способ - to_ Pandas () или collect (). Оба могут вызвать проблемы при большом размере данных. Существует множество распределенных пакетов глубокого обучения, которые обеспечивают встроенную поддержку фреймов данных, поэтому вы можете запускать свою модель напрямую. Проверьте ресурсы здесь:

https://joerihermans.com/work/distributed-keras/

https://towardsdatascience.com/distributed-deep-learning-pipelines-with-pyspark-and-keras-a3a1c22b9239

Их использование будет лучшим способом go.

Answer 2

Вы можете выполнить основной расчет на фрейме данных, как показано ниже:

from pyspark.sql.functions import *
df = spark.read.option('header', 'true').csv('data/d.csv').select("customer",col("A").cast("float"),col("B").cast("float"),col("C").cast("float"))
df.show()

+--------+--------+--------+--------+
|customer|       A|       B|       C|
+--------+--------+--------+--------+
|       1|0.999471|0.746963|0.098438|
|       1|0.986164|0.762772| 0.71535|
|       1|0.656797|0.975615|0.002074|
|       2|0.892825|0.542348|0.648907|
|       2|0.806137|0.806392|0.365054|
|       2|0.707369|0.535159|0.205938|
|       3|0.072516|0.381937|0.076068|
|       3|0.442076|0.634749| 0.91852|
|       3|0.628006|0.842492|0.971928|
+--------+--------+--------+--------+

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

df2 = np.array(df.groupBy("customer").agg(collect_list(array(array("A"), array("B"), array("C")))).collect())
# select data from customer = 1
images = np.asarray([x[1] for x in df2 if x[0] == '1' ])
print images

вывод:

[[[[0.99947101]
   [0.74696302]
   [0.098438  ]]

  [[0.98616397]
   [0.76277202]
   [0.71534997]]

  [[0.65679699]
   [0.97561502]
   [0.002074  ]]]]