La transposition de la colonne à la ligne avec Spark

L'Étincelle local de l'algèbre linéaire, les bibliothèques sont actuellement très faible et ils ne comprennent pas les opérations de base comme ci-dessus.

Il y a un JIRA pour la fixation de cette pour Spark 2.1 - mais ce ne sera pas vous aider à aujourd'hui.

Quelque chose à considérer: l'exécution d'une transposition nécessitera probablement complètement brouiller les données.

De maintenant, vous aurez besoin d'écrire RDD code directement. J'ai écrit transpose en scala - mais pas en python. Voici la scala version:

 def transpose(mat: DMatrix) = {
    val nCols = mat(0).length
    val matT = mat
      .flatten
      .zipWithIndex
      .groupBy {
      _._2 % nCols
    }
      .toSeq.sortBy {
      _._1
    }
      .map(_._2)
      .map(_.map(_._1))
      .toArray
    matT
  }

De sorte que vous pouvez convertir en python pour votre utilisation. Je n'ai pas de bande passante d'écrire/de test à ce moment-là: laissez-moi savoir si vous avez été incapable de faire la conversion.

Au moins les éléments suivants sont facilement convertis à python.

• zipWithIndex --> enumerate() (python équivalent - crédit @zero323)
• map --> [someOperation(x) for x in ..]
• groupBy --> itertools.groupBy()

Ici est la mise en œuvre de flatten qui n'ont pas d'équivalent python:

  def flatten(L):
        for item in L:
            try:
                for i in flatten(item):
                    yield i
            except TypeError:
                yield item

De sorte que vous devriez être capable de mettre ensemble pour trouver une solution.

OriginalL'auteur javadba

J'ai pris la Scala réponse @javadba a écrit et créé une version de Python pour la transposition de toutes les colonnes dans une DataFrame. Cela pourrait être un peu différent de ce que l'OP a demander...

from itertools import chain
from pyspark.sql import DataFrame


def _sort_transpose_tuple(tup):
    x, y = tup
    return x, tuple(zip(*sorted(y, key=lambda v_k: v_k[1], reverse=False)))[0]


def transpose(X):
    """Transpose a PySpark DataFrame.

    Parameters
    ----------
    X : PySpark ``DataFrame``
        The ``DataFrame`` that should be tranposed.
    """
    # validate
    if not isinstance(X, DataFrame):
        raise TypeError('X should be a DataFrame, not a %s' 
                        % type(X))

    cols = X.columns
    n_features = len(cols)

    # Sorry for this unreadability...
    return X.rdd.flatMap( # make into an RDD
        lambda xs: chain(xs)).zipWithIndex().groupBy( # zip index
        lambda val_idx: val_idx[1] % n_features).sortBy( # group by index % n_features as key
        lambda grp_res: grp_res[0]).map( # sort by index % n_features key
        lambda grp_res: _sort_transpose_tuple(grp_res)).map( # maintain order
        lambda key_col: key_col[1]).toDF() # return to DF

Par exemple:

>>> X = sc.parallelize([(1,2,3), (4,5,6), (7,8,9)]).toDF()
>>> X.show()
+---+---+---+
| _1| _2| _3|
+---+---+---+
|  1|  2|  3|
|  4|  5|  6|
|  7|  8|  9|
+---+---+---+

>>> transpose(X).show()
+---+---+---+
| _1| _2| _3|
+---+---+---+
|  1|  4|  7|
|  2|  5|  8|
|  3|  6|  9|
+---+---+---+

OriginalL'auteur Tgsmith61591

Une façon de résoudre avec pyspark sql à l'aide de fonctions create_map et explode.

from pyspark.sql import functions as func
#Use `create_map` to create the map of columns with constant 
df = df.withColumn('mapCol', \
                    func.create_map(func.lit('col_1'),df.col_1,
                                    func.lit('col_2'),df.col_2,
                                    func.lit('col_3'),df.col_3
                                   ) 
                  )
#Use explode function to explode the map 
res = df.select('*',func.explode(df.mapCol).alias('col_id','col_value'))
res.show()

OriginalL'auteur Vamsi Prabhala