Auswählen zufällige Elemente aus einer Funke GroupedData Objekt

Gut, es ist irgendwie falsch. GroupedData ist nicht wirklich entwickelt, für die ein Zugriff auf Daten. Es beschreibt nur die Gruppierung der Kriterien und bietet aggregationsmethoden. Siehe meine Antwort auf Mit groupBy-Zündkerzen und immer wieder zu einem DataFrame für mehr details.

Ein weiteres problem mit dieser Idee ist die Auswahl N random samples. Es ist eine Aufgabe, die schwer zu erreichen ist parallel ohne psychische Gruppierung von Daten, und es ist nicht etwas, das passiert, wenn man call groupBy auf eine DataFrame:

Gibt es mindestens zwei Möglichkeiten, damit umzugehen:

• convert to RDD, groupBy und Durchführung der lokalen sampling -

  import random
  
  n = 3
  
  def sample(iter, n): 
      rs = random.Random()  # We should probably use os.urandom as a seed
      return rs.sample(list(iter), n)    
  
  df = sqlContext.createDataFrame(
      [(x, y, random.random()) for x in (1, 2, 3) for y in "abcdefghi"], 
      ("teamId", "x1", "x2"))
  
  grouped = df.rdd.map(lambda row: (row.teamId, row)).groupByKey()
  
  sampled = sqlContext.createDataFrame(
      grouped.flatMap(lambda kv: sample(kv[1], n)))
  
  sampled.show()
  
  ## +------+---+-------------------+
  ## |teamId| x1|                 x2|
  ## +------+---+-------------------+
  ## |     1|  g|   0.81921738561455|
  ## |     1|  f| 0.8563875814036598|
  ## |     1|  a| 0.9010425238735935|
  ## |     2|  c| 0.3864428179837973|
  ## |     2|  g|0.06233470405822805|
  ## |     2|  d|0.37620872770129155|
  ## |     3|  f| 0.7518901502732027|
  ## |     3|  e| 0.5142305439671874|
  ## |     3|  d| 0.6250620479303716|
  ## +------+---+-------------------+

• window-Funktionen

  from pyspark.sql import Window
  from pyspark.sql.functions import col, rand, rowNumber
  
  w = Window.partitionBy(col("teamId")).orderBy(col("rnd_"))
  
  sampled = (df
      .withColumn("rnd_", rand())  # Add random numbers column
      .withColumn("rn_", rowNumber().over(w))  # Add rowNumber over windw
      .where(col("rn_") <= n)  # Take n observations
      .drop("rn_")  # drop helper columns
      .drop("rnd_"))
  
  sampled.show()
  
  ## +------+---+--------------------+
  ## |teamId| x1|                  x2|
  ## +------+---+--------------------+
  ## |     1|  f|  0.8563875814036598|
  ## |     1|  g|    0.81921738561455|
  ## |     1|  i|  0.8173912535268248|
  ## |     2|  h| 0.10862995810038856|
  ## |     2|  c|  0.3864428179837973|
  ## |     2|  a|  0.6695356657072442|
  ## |     3|  b|0.012329360826023095|
  ## |     3|  a|  0.6450777858109182|
  ## |     3|  e|  0.5142305439671874|
  ## +------+---+--------------------+

aber ich fürchte, beides wird sehr teuer sein. Wenn die Größe der einzelnen Gruppen ist ausgewogen und relativ groß, ich würde einfach verwenden DataFrame.randomSplit.

Wenn die Anzahl der Gruppen ist relativ klein, ist es möglich, zu versuchen, etwas anderes:

from pyspark.sql.functions import count, udf
from pyspark.sql.types import BooleanType
from operator import truediv

counts = (df
    .groupBy(col("teamId"))
    .agg(count("*").alias("n"))
    .rdd.map(lambda r: (r.teamId, r.n))
    .collectAsMap()) 

# This defines fraction of observations from a group which should
# be taken to get n values 
counts_bd = sc.broadcast({k: truediv(n, v) for (k, v) in counts.items()})

to_take = udf(lambda k, rnd: rnd <= counts_bd.value.get(k), BooleanType())

sampled = (df
    .withColumn("rnd_", rand())
    .where(to_take(col("teamId"), col("rnd_")))
    .drop("rnd_"))

sampled.show()

## +------+---+--------------------+
## |teamId| x1|                  x2|
## +------+---+--------------------+
## |     1|  d| 0.14815204548854788|
## |     1|  f|  0.8563875814036598|
## |     1|  g|    0.81921738561455|
## |     2|  a|  0.6695356657072442|
## |     2|  d| 0.37620872770129155|
## |     2|  g| 0.06233470405822805|
## |     3|  b|0.012329360826023095|
## |     3|  h|  0.9022527556458557|
## +------+---+--------------------+

Spark 1.5+ können Sie ersetzen udf mit einem Aufruf sampleBy Methode:

df.sampleBy("teamId", counts_bd.value)

Wird es nicht geben Ihnen genaue Anzahl der Beobachtungen sollte aber gut genug sein, die meisten der Zeit, die so lange als Anzahl der Beobachtungen pro Gruppe ist groß genug, um die richtige Muster. Sie können auch sampleByKey auf einem RDD in ähnlicher Weise.

InformationsquelleAutor zero323