Matrices en 3 colonnes

Représentation d’une matrice avec Spark / Map / Reduce.

Ce notebook propose d’implémenter un produit matriciel sous Spark. Spark comme SQL n’aime pas trop avoir un nombre de colonnes variables. La première étape consiste à transformer les matrices I\times J en tableau de trois colonnes (i,j,coefficient).

Session spark with no cluster

Spark est censé tourner sur un cluster. Mais ce n’est pas essentielle pour comprendre la logique. Le notebook tourne donc en local.

[16]:

from pyspark.sql import SparkSession

# Create a SparkSession
spark = SparkSession.builder.appName("MySparkApp").master("local[*]").getOrCreate()
sc = spark.sparkContext

Création d’une matrice aléatoire

[17]:

from numpy.random import rand

rnd1 = rand(10, 10)
rnd2 = rand(10, 2)
rnd1 @ rnd2

[17]:

array([[1.2962659 , 2.75236533],
       [2.26735872, 2.89961464],
       [1.29025917, 2.34056096],
       [1.82876448, 3.42098919],
       [1.91448985, 3.37298335],
       [1.84269033, 1.98821207],
       [2.28212544, 3.05316399],
       [1.88631937, 3.06186776],
       [2.67976259, 3.61823182],
       [1.70446473, 2.71078996]])

[18]:

import pandas

df1 = pandas.DataFrame(rnd1)
df2 = pandas.DataFrame(rnd2)
df2

[18]:
0 1
0 0.425791 0.508217
1 0.444969 0.926192
2 0.078127 0.349568
3 0.707894 0.845050
4 0.179368 0.555457
5 0.571995 0.419750
6 0.782654 0.712389
7 0.165768 0.830360
8 0.043705 0.759277
9 0.502934 0.110957 
[19]:

df1.to_csv("rnd1.txt", sep="\t", header=None, index=False)
df2.to_csv("rnd2.txt", sep="\t", header=None, index=False)

[20]:

mat1 = spark.sparkContext.textFile("rnd1.txt")

Conversion d’une matrice au format Spark

Lorsqu’un traitement est distribué en Map/Reduce, il n’est pas possible de s’appuyer sur l’ordre dans lequel sont traitées les lignes. Le plus est d’ajouter cette information sur chaque ligne plutôt que de chercher à la récupérer.

[21]:

df1.to_csv("rnd1.txt", sep="\t", header=None, index=True)
df2.to_csv("rnd2.txt", sep="\t", header=None, index=True)

[22]:

def process_mat_row(row):
    values = row.split("\t")
    index = int(values[0])
    values = [float(_) for _ in values[1:]]
    return [[index, j, v] for j, v in enumerate(values)]

[23]:

mat1 = sc.textFile("rnd1.txt")
new_mat1 = mat1.flatMap(process_mat_row)
new_mat1.take(12)

[23]:

[[0, 0, 0.03925624285714491],
 [0, 1, 0.17938143471086276],
 [0, 2, 0.9271561615741587],
 [0, 3, 0.13497740334270003],
 [0, 4, 0.20256379268518632],
 [0, 5, 0.15623553510014287],
 [0, 6, 0.7926007655892027],
 [0, 7, 0.9953375114509172],
 [0, 8, 0.7100110433596362],
 [0, 9, 0.1783301416123766],
 [1, 0, 0.6661128314162409],
 [1, 1, 0.16617620104743758]]

[24]:

mat2 = sc.textFile("rnd2.txt")
new_mat2 = mat2.flatMap(process_mat_row)
new_mat2.take(12)

[24]:

[[0, 0, 0.4257910374269508],
 [0, 1, 0.5082167960207694],
 [1, 0, 0.4449691717763494],
 [1, 1, 0.9261922479892456],
 [2, 0, 0.07812708982401129],
 [2, 1, 0.34956810336320765],
 [3, 0, 0.7078936713530861],
 [3, 1, 0.8450500475013194],
 [4, 0, 0.17936816503487407],
 [4, 1, 0.5554570108793752],
 [5, 0, 0.5719951163381093],
 [5, 1, 0.41975047374547725]]

Produit matriciel

Il faut d’abord faire la jointure avec la méthode join. Il faut que la clé soit sur la première colonne.

[25]:

def key_ij(row):
    return row[0], (row[1], row[2])


def key_ji(row):
    return row[1], (row[0], row[2])


mat_join = new_mat1.map(key_ji).join(new_mat2.map(key_ij))
mat_join.take(12)

[25]:

[(0, ((0, 0.03925624285714491), (0, 0.4257910374269508))),
 (0, ((0, 0.03925624285714491), (1, 0.5082167960207694))),
 (0, ((1, 0.6661128314162409), (0, 0.4257910374269508))),
 (0, ((1, 0.6661128314162409), (1, 0.5082167960207694))),
 (0, ((2, 0.8809481084845031), (0, 0.4257910374269508))),
 (0, ((2, 0.8809481084845031), (1, 0.5082167960207694))),
 (0, ((3, 0.09489201762862454), (0, 0.4257910374269508))),
 (0, ((3, 0.09489201762862454), (1, 0.5082167960207694))),
 (0, ((4, 0.9071785348453754), (0, 0.4257910374269508))),
 (0, ((4, 0.9071785348453754), (1, 0.5082167960207694))),
 (0, ((5, 0.8535870359856828), (0, 0.4257910374269508))),
 (0, ((5, 0.8535870359856828), (1, 0.5082167960207694)))]

On effectue le produit matriciel.

[26]:

def produit_matriciel(row):
    index, ((i, v1), (j, v2)) = row
    return i, j, v1 * v2


produit = mat_join.map(produit_matriciel)
produit.take(12)

[26]:

[(0, 0, 0.016714956371628058),
 (0, 1, 0.019950681968671398),
 (1, 0, 0.2836248735321248),
 (1, 1, 0.33852972897068484),
 (2, 0, 0.37509980903092655),
 (2, 1, 0.44771262515455135),
 (3, 0, 0.04040417062962855),
 (3, 1, 0.04822571716716593),
 (4, 0, 0.3862684894832736),
 (4, 1, 0.4610433683979326),
 (5, 0, 0.3634497095865398),
 (5, 1, 0.4338072685535089)]

Il ne reste plus qu’à agréger reduceByKey. La documentation fournit un exemple facilement transposable. Elle indique aussi : Merge the values for each key using an associative and commutative reduce function. Pourquoi précise-t-elle associative et commutative ? Cela signifie que le résultat ne dépend pas de l’ordre dans lequel l’agrégation est réalisée et qu’on peut commencer à agréger sans attendre d’avoir regroupé toutes les valeurs associées à une clé.

  • Cas 1 : groupBy + agrégation qui commence une fois les valeurs regroupées

  • Cas 2 : reduceByKey + agrégation qui commence dès les premières valeurs regroupées

Le cas 2 est moins consommateur en terme de données. Le cas 1 n’est possible que si les valeurs agrégées ne sont pas trop nombreuses. Ca tombe bien, dans notre cas, le cas 2 convient.

[27]:

from operator import add

final = produit.map(lambda row: ((row[0], row[1]), row[2])).reduceByKey(add)
aslist = final.collect()
aslist.sort()
aslist

[27]:

[((0, 0), 1.2962658962226397),
 ((0, 1), 2.7523653340144056),
 ((1, 0), 2.267358716094168),
 ((1, 1), 2.899614641397404),
 ((2, 0), 1.2902591694424805),
 ((2, 1), 2.3405609608679425),
 ((3, 0), 1.8287644824176785),
 ((3, 1), 3.420989188235977),
 ((4, 0), 1.9144898451263708),
 ((4, 1), 3.372983349186469),
 ((5, 0), 1.842690333964681),
 ((5, 1), 1.9882120730740667),
 ((6, 0), 2.282125435665258),
 ((6, 1), 3.053163988860857),
 ((7, 0), 1.8863193688285897),
 ((7, 1), 3.061867764510199),
 ((8, 0), 2.6797625884756293),
 ((8, 1), 3.6182318180423017),
 ((9, 0), 1.7044647282526524),
 ((9, 1), 2.710789957326838)]

Résultat initial :

[28]:

rnd1 @ rnd2

[28]:

array([[1.2962659 , 2.75236533],
       [2.26735872, 2.89961464],
       [1.29025917, 2.34056096],
       [1.82876448, 3.42098919],
       [1.91448985, 3.37298335],
       [1.84269033, 1.98821207],
       [2.28212544, 3.05316399],
       [1.88631937, 3.06186776],
       [2.67976259, 3.61823182],
       [1.70446473, 2.71078996]])

Même algorithme avec les Spark DataFrame

On a besoin de réaliser un flatMap. Une façon de faire est de créer des colonnes qui sont de type composé : un tableau, une structure. La multiplication des lignes est obtenue avec la fonction explode.

[29]:

schema = ["index"] + ["c%d" % i for i in range(1, 11)]
mat1 = spark.createDataFrame(
    pandas.read_csv("rnd1.txt", header=None, sep="\t"), schema=schema
)

/home/xadupre/.local/lib/python3.10/site-packages/pyspark/sql/pandas/conversion.py:485: FutureWarning: is_datetime64tz_dtype is deprecated and will be removed in a future version. Check `isinstance(dtype, pd.DatetimeTZDtype)` instead.
  if should_localize and is_datetime64tz_dtype(s.dtype) and s.dt.tz is not None:

[30]:

mat1.printSchema()

root
 |-- index: long (nullable = true)
 |-- c1: double (nullable = true)
 |-- c2: double (nullable = true)
 |-- c3: double (nullable = true)
 |-- c4: double (nullable = true)
 |-- c5: double (nullable = true)
 |-- c6: double (nullable = true)
 |-- c7: double (nullable = true)
 |-- c8: double (nullable = true)
 |-- c9: double (nullable = true)
 |-- c10: double (nullable = true)


[31]:

schema = ["index"] + ["c%d" % i for i in range(1, 3)]
mat2 = spark.createDataFrame(
    pandas.read_csv("rnd2.txt", header=None, sep="\t"), schema=schema
)

/home/xadupre/.local/lib/python3.10/site-packages/pyspark/sql/pandas/conversion.py:485: FutureWarning: is_datetime64tz_dtype is deprecated and will be removed in a future version. Check `isinstance(dtype, pd.DatetimeTZDtype)` instead.
  if should_localize and is_datetime64tz_dtype(s.dtype) and s.dt.tz is not None:

[32]:

mat2.printSchema()

root
 |-- index: long (nullable = true)
 |-- c1: double (nullable = true)
 |-- c2: double (nullable = true)

Nous allons avoir besoin de quelques-uns des fonctions et types suivant :

Je recommande le type FloatType qui prend deux fois moins de place pour une précision moindre mais suffisante dans la plupart des cas.

[33]:

from pyspark.sql.types import (
    ArrayType,
    StructField,
    StructType,
    DoubleType,
    IntegerType,
)
from pyspark.sql.functions import explode, posexplode, array
from pyspark.sql import Row

[34]:

cols = ["c%d" % i for i in range(1, 11)]
mat1_array = mat1.select(mat1.index, array(*cols).alias("x"))
mat1_array.printSchema()

root
 |-- index: long (nullable = true)
 |-- x: array (nullable = false)
 |    |-- element: double (containsNull = true)


[35]:

mat1_exploded = mat1_array.select("index", posexplode("x"))
mat1_exploded.printSchema()

root
 |-- index: long (nullable = true)
 |-- pos: integer (nullable = false)
 |-- col: double (nullable = true)


[36]:

mat1.toPandas().shape, mat1_exploded.toPandas().shape

[36]:

((10, 11), (100, 3))

On recommence le même procédé pour l’autre matrice.

[37]:

cols = ["c%d" % i for i in range(1, 3)]
mat2_array = mat2.select(mat2.index, array(*cols).alias("x"))
mat2_exploded = mat2_array.select("index", posexplode("x"))

Il ne reste plus qu’à faire le produit avec la méthode join après avoir renommé les colonnes avant la jointure pour éviter les ambiguïtés.

[38]:

mat2_exp2 = (
    mat2_exploded.withColumnRenamed("index", "index2")
    .withColumnRenamed("pos", "pos2")
    .withColumnRenamed("col", "col2")
)
produit = mat1_exploded.join(mat2_exp2, mat1_exploded.pos == mat2_exp2.index2)

[39]:

produit.printSchema()

root
 |-- index: long (nullable = true)
 |-- pos: integer (nullable = false)
 |-- col: double (nullable = true)
 |-- index2: long (nullable = true)
 |-- pos2: integer (nullable = false)
 |-- col2: double (nullable = true)


[40]:

produit.toPandas().head()



[40]:
index pos col index2 pos2 col2
0 0 0 0.039256 0 0 0.425791
1 0 0 0.039256 0 1 0.508217
2 1 0 0.666113 0 0 0.425791
3 1 0 0.666113 0 1 0.508217
4 2 0 0.880948 0 0 0.425791 
[41]:

prod = produit.select(
    produit.index.alias("i"),
    produit.pos2.alias("j"),
    (produit.col * produit.col2).alias("val"),
)
final = prod.groupby("i", "j").sum("val")

[42]:

final.printSchema()

root
 |-- i: long (nullable = true)
 |-- j: integer (nullable = false)
 |-- sum(val): double (nullable = true)


[43]:

df = final.toPandas()



[44]:

df.sort_values(["i", "j"]).head()

[44]:
i j sum(val)
7 0 0 1.296266
10 0 1 2.752365
18 1 0 2.267359
3 1 1 2.899615
6 2 0 1.290259 
[45]:

df.shape

[45]:

(20, 3)

Fin

[ ]:

spark.stop()

Notebook on github