PySpark DataFrames - façon d'énumérer sans convertir en Pandas?

Cela ne fonctionne pas car:

1. le deuxième argument pour withColumn devrait être un Column n'est pas une collection. np.array ne marchera pas ici
2. lorsque vous passez "index in indexes" comme une expression SQL à where indexes est hors de portée et il n'est pas résolu comme un identificateur valide

PySpark >= 1.4.0

Vous pouvez ajouter des numéros de ligne à l'aide de fenêtre correspondante de la fonction et de la requête à l'aide de Column.isin méthode ou un formatage de chaîne de requête:

from pyspark.sql.functions import col, rowNumber
from pyspark.sql.window import Window

w = Window.orderBy()
indexed = df.withColumn("index", rowNumber().over(w))

# Using DSL
indexed.where(col("index").isin(set(indexes)))

# Using SQL expression
indexed.where("index in ({0})".format(",".join(str(x) for x in indexes)))

On dirait les fonctions de la fenêtre appelée sans PARTITION BY clause de déplacer toutes les données de la partition unique, ce qui est en haut est peut-être pas la meilleure solution, après tout.

Importe quel plus rapide et la plus simple façon de traiter avec elle?

Pas vraiment. Spark DataFrames ne prennent pas en charge aléatoire de l'accès aux lignes.

PairedRDD peut être consulté à l'aide de lookup méthode qui est relativement rapide si les données sont partitionnées en utilisant HashPartitioner. Il est également indexées ca projet qui contribue à l'efficacité des recherches.

Modifier:

Indépendant de PySpark version, vous pouvez essayer quelque chose comme cela:

from pyspark.sql import Row
from pyspark.sql.types import StructType, StructField, LongType

row = Row("char")
row_with_index = Row("char", "index")

df = sc.parallelize(row(chr(x)) for x in range(97, 112)).toDF()
df.show(5)

## +----+
## |char|
## +----+
## |   a|
## |   b|
## |   c|
## |   d|
## |   e|
## +----+
## only showing top 5 rows

# This part is not tested but should work and save some work later
schema  = StructType(
    df.schema.fields[:] + [StructField("index", LongType(), False)])

indexed = (df.rdd # Extract rdd
    .zipWithIndex() # Add index
    .map(lambda ri: row_with_index(*list(ri[0]) + [ri[1]])) # Map to rows
    .toDF(schema)) # It will work without schema but will be more expensive

# inSet in Spark < 1.3
indexed.where(col("index").isin(indexes))

Vous pouvez certainement ajouter un tableau pour l'indexation, un tableau de votre choix en effet:
En Scala, nous devons d'abord créer une indexation de Tableau:

val index_array=(1 to df.count.toInt).toArray

index_array: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

Vous pouvez maintenant ajouter cet article à votre DF. Tout d'abord, Pour cela, vous devez ouvrir nos DF et l'obtenir comme un tableau, puis zip avec votre index_array et puis nous convertir le nouveau tableau et RDD. La dernière étape consiste à obtenir que d'un DF:

final_df = sc.parallelize((df.collect.map(
    x=>(x(0),x(1))) zip index_array).map(
    x=>(x._1._1.toString,x._1._2.toString,x._2))).
    toDF("column_name")

L'indexation serait plus clair après.