Manipulation des données PySpark

Cet article résume les fonctionnalités et les opérations sur les données de PySpark.

À propos de PySpark

Caractéristiques de PySpark (Spark)

• Entrée / sortie de fichier
• Entrée: un seul fichier est acceptable
• Sortie: le nom du fichier de sortie ne peut pas être attribué (seul le nom du dossier peut être spécifié). Sortie dans plusieurs fichiers directement sous le dossier spécifié
• Évaluation du retard
• Le traitement est exécuté lorsque le fichier est sorti ou le résultat est sorti.
• Normalement, seul le plan d'exécution est calculé

«Partitionnement» et «compartimentage»

Un concept important de ʻApache Hive` dans le fonctionnement de PySpark

• Partitionnement: divisez la destination de sortie des fichiers en dossiers. Vous pouvez limiter la plage de fichiers à lire.
• Bucketing: subdivision des données dans un fichier avec une fonction de hachage. Il peut être lu efficacement.

Pour plus d'informations sur le partitionnement et le compartimentage, voir ici.

Vérification de l'état d'utilisation des ressources informatiques

Si vos données sont lentes, il est judicieux d'utiliser Ganglia pour voir comment vos ressources de calcul sont utilisées.

En particulier, le volume de communication réseau (= volume de transfert de données) est faible et le traitement prend souvent du temps. Dans ce cas, vous pourrez peut-être le résoudre en prenant les mesures suivantes.

• Lire les fichiers distribués plutôt que des fichiers uniques
• En raison du retard d'évaluation, le résultat de l'exécution de nombreux processus est lent, utilisez donc l'une des méthodes suivantes
• Utilisez le cache de trame de données: par exemple df = df.cache ()
• Crachez une fois dans un dossier, relisez le résultat de sortie et exécutez le traitement suivant

Extrait de code PySpark

Les variables suivantes sont supposées avoir été générées.

• spark: spark context
• path: quelque chose de chemin de fichier
• Éléments ʻimport` dans la section suivante

Mise en garde

1. Le contenu affiché par > df.show () peut ne pas toujours être correct dans le but de saisir l'image.
2. Vous pouvez soudainement utiliser une variable autre que celles listées ci-dessus.
3. Le chemin commence à «s3: //» car il est censé s'exécuter sur AWS.
4. Nous prévoyons d'ajouter et de modifier séquentiellement.
5. ** Si vous faites une erreur de syntaxe ou d'écriture, veuillez laisser un commentaire. ** **

import

Les éléments suivants sont principalement importés lors de l'utilisation de «spark».

# from pyspark.sql.functions import *Dans certains cas,
#J'aime spécifier l'espace de noms de la fonction avec F car c'est plus facile à comprendre.
#Cependant, F viole PEP8. .. ..
from pyspark.sql import functions as F
from pyspark.sql.types import FloatType, TimestampType, StringType
from pyspark.sql.window import Window

Paramètres de l'environnement d'exécution

• Lorsque vous utilisez EMR sur AWS, définissez sur JST car l'heure de l'instance est ʻUTC`.

spark.conf.set("spark.sql.session.timeZone", "Asia/Tokyo")

initialize spark Ce n'est pas nécessaire sur JupyterHub de ʻEMR`, mais lors de l'exécution avec un script python, L'initialisation de l'instance de «spark» est requise.

# spark initialization
spark = SparkSession.builder.appName("{your app name here}").getOrCreate()

Lecture des données

• Lors de la lecture à partir d'une chaîne de caractères

df = spark.read.parquet(path)

• Plusieurs fichiers / dossiers peuvent être lus à la fois en utilisant *

# dt=2020-01-01/Lisez tous les fichiers ci-dessous
df = spark.read.parquet("s3://some-bucket/data/dt=2020-01-01/*.parquet")

# dt=2020-01-01/De dt=2020-01-31/Lisez tous les fichiers ci-dessous
df = spark.read.parquet("s3://some-bucket/data/dt=2020-01-*/*.parquet")

• from list : Ceci est également possible si vous souhaitez étendre plusieurs dossiers

#lit les fichiers dans les chemins inclus dans la liste des chemins
df = spark.read.parquet(*paths)

• Si vous utilisez la partition et que vous souhaitez la mettre dans la colonne du bloc de données qui lit la partition

#Ajouter une partition à la colonne et lire
df = spark.read.option("basePath", parent_path).parquet(*paths)

• Trame de données de cache

En sauvegardant le résultat de l'évaluation du retard dans la mémoire, un traitement à grande vitesse devient possible. Il est préférable de mettre en cache () les données fréquemment utilisées et de les utiliser surtout après le traitement.

#Cache en mémoire
df = df.cache()
#Ou
#Cache en mémoire par défaut, la destination du cache peut être changée en stockage, etc. avec un argument facultatif
df = df.persist()

• Confirmation du type de données

df.printSchema()

root
 |-- id: string (nullable = true)
 |-- name: string (nullable = true)

Sortie de données

#csv (l'en-tête n'est pas donné dans ce cas)
df.write.csv(path)

# parquet
df.write.parquet(path)

• en-tête: Remarque uniquement pour csv

#Dans le cas de csv, il n'est donné que si le paramètre de sortie de l'en-tête est défini.
df.write.mode("overwrite").option("header", "True").csv(path)
# or
df.write.mode("overwrite").csv(path, header=True)
#Dans le cas du parquet, il est édité par défaut même si l'en-tête n'est pas spécifié.
df.write.parquet(path)

• compression: compression

# gzip with csv
df.write.csv(path, compression="gzip")

#accrocheur avec du parquet (devrait-il être compressé par défaut?)
df.write.option("compression", "snappy").parquet(path)

• partitionBy: lorsque vous souhaitez partitionner par le nom de colonne de la trame de données lors de la sortie

Dans le cas de l'exemple suivant, il sera affiché dans le dossier / dt = {dt_col} / count = {count_col} / {file} .parquet.

df.repartition("dt", "count").write.partitionBy("dt", "count").parqeut(path)

• coalesce: le contenu qui est normalement sorti dans plusieurs fichiers peut être sorti ensemble dans un seul fichier

Si vous effectuez une fusion après plusieurs processus, la vitesse de traitement ralentira, donc si possible, il est préférable de sortir le fichier normalement, puis de fusionner à nouveau la lecture.

#Peut être lent après plusieurs processus
df.coalesce(1).write.csv(path, header=True)

#Recommandé si possible (sortie → lecture → sortie)
df.write.parquet(path)
alt_df = spark.read.parquet(path)
alt_df.coalesce(1).write.csv(path, header=True)

• repartition: spécifiez le nombre de divisions du fichier de sortie

df.repartition(20).write.parquet(path)

• write.mode (): méthode de sortie sélectionnable

# write.mode()Arguments pouvant être utilisés dans'overwrite', 'append', 'ignore', 'error', 'errorifexists'
#J'utilise souvent l'écrasement
#Normalement, une erreur se produit si le fichier existe dans le dossier de destination de sortie.
df.write.parquet(path)

#Si vous voulez écraser
df.write.mode("overwrite").parquet(path)

#Si vous souhaitez ajouter au dossier actuel
df.write.mode("append").parquet(path)

Génération de trames de données

Il s'agit d'une méthode de création d'un bloc de données par programme, et non à partir de la lecture d'un fichier.

• Pour une seule colonne

#Créer un bloc de données à une seule colonne
id_list = ["A001", "A002", "B001"]
df = spark.createDataFrame(id_list, StringType()).toDF("id")

• Pour plusieurs colonnes

#L'élément à l'intérieur est tuple,Spécifiez enfin le nom de la colonne
df = spark.createDataFrame([
    ("a", None, None),
    ("a", "code1", None),
    ("a", "code2", "name2"),
], ["id", "code", "name"])

> df.show()
+---+-----+-----+
| id| code| name|
+---+-----+-----+
|  a| null| null|
|  a|code1| null|
|  a|code2|name2|
+---+-----+-----+

# =======================
#Lors de la création à l'aide de rdd une fois
rdd = sc.parallelize(
    [
        (0, "A", 223, "201603", "PORT"), 
        (0, "A", 22, "201602", "PORT"), 
        (0, "A", 422, "201601", "DOCK"), 
        (1, "B", 3213, "201602", "DOCK"), 
        (1, "B", 3213, "201601", "PORT"), 
        (2, "C", 2321, "201601", "DOCK")
    ]
)
df_data = spark.createDataFrame(rdd, ["id","type", "cost", "date", "ship"])

> df.show()
+---+----+----+------+----+
| id|type|cost|  date|ship|
+---+----+----+------+----+
|  0|   A| 223|201603|PORT|
|  0|   A|  22|201602|PORT|
|  0|   A| 422|201601|DOCK|
|  1|   B|3213|201602|DOCK|
|  1|   B|3213|201601|PORT|
|  2|   C|2321|201601|DOCK|
+---+----+----+------+----+

Ajouter une colonne (withColumn ())

Dans PySpark, l'analyse est souvent effectuée en utilisant le "processus d'ajout d'une nouvelle colonne".

# new_col_Créez une nouvelle colonne appelée nom et donnez-lui une valeur littérale (= constante) de 1.
df = df.withColumn("new_col_name", F.lit(1))

• F.input_file_name (): Récupère le nom du fichier lu

#Donnez le chemin du fichier lu
df = df.withColumn("file_path", F.input_file_name())

#Obtenez le nom du fichier à partir du chemin du fichier lu
df = df.withColumn("file_name", F.split(col("file_path"), "/").getItem({int:Dernière valeur d'index}))

• cast (): conversion de type

#Spécifié par chaîne de caractères
df = df.withColumn("total_count", F.col("total_count").cast("double"))

#Spécifié par les types PySpark
df = df.withColumn("value", F.lit("1").cast(StringType()))

• F.when (). Sinon (): Modifiez la valeur à ajouter en fonction de la condition

#Si vous souhaitez ajouter une colonne de valeurs selon les conditions
# F.when(condtion, value).otherwise(else_value)
df = df.withColumn("is_even", F.when(F.col("number") % 2 == 0, 1).otherwise(0))

#En cas de conditions multiples
df = df.withColumn("search_result", F.when( (F.col("id") % 2 == 0) & (F.col("room") % 2 == 0), 1).otherwise(0))

• isNotNull (): détermine si nul

df = df.withColumn("is_touched", F.col("value").isNotNull())

• F.regexp_replace (): Remplacement de caractères à l'aide d'expressions régulières

df = df.withColumn("replaced_id", F.regexp_replace(F.col("id"), "A", "C"))

• Opérations liées au temps

# date time -> epoch time
df = df.withColumn("epochtime", F.unix_timestamp(F.col("timestamp"), "yyyy-MM-dd HH:mm:ssZ"))

# epoch time -> date time
# 1555259647 -> 2019-04-14 16:34:07
df = df.withColumn("datetime", F.to_timestamp(df["epochtime"]))

# datetime -> string
# 2019-04-14 16:34:07 -> 2019-04-14
string_format = "yyyy-MM-dd"
df = df.withColumn("dt", F.date_format(F.col("datetime"), string_format))

# epoch time:Une chaîne de nombres d'environ 10 chiffres. Nombre de secondes depuis le 1er janvier 1970
df = df.withColumn("hour", F.hour(F.col("epochtime")))
df = df.withColumn("hour", F.hour(F.col("timestamp")))

#Tronquer datetime à la largeur de temps spécifiée
df = df.withColumn("hour", F.date_trunc("hour", "datetime"))
df = df.withColumn("week", F.date_trunc("week", "datetime"))

Il existe de nombreuses autres fonctions disponibles dans Takutan withColumn. Veuillez également consulter le site de référence.

Combiner des trames de données

La méthode pour joindre deux DataFrames horizontalement / verticalement est join () / union ().

#Spécifiez les colonnes à rejoindre sur
df = left_df.join(right_df, on="id")

# data-Pour différentes colonnes pour chaque cadre
df = left_df.join(right_df, left_df.id_1 == right_df.id_2)

#Vous pouvez également spécifier la méthode de combinaison
# how:= inner, left, right, left_semi, left_anti, cross, outer, full, left_outer, right_outer
df = left_df.join(right_df, on="id", how="inner")

• Rejoignez avec plusieurs colonnes

df = left_df.join(right_df, on=["id", "dt"])

• F.broadcast () join: Comment distribuer et joindre efficacement des données à chaque cluster
• L'efficacité augmente (peut augmenter) lorsqu'elle est utilisée lorsque la taille des données de chaque trame de données est déséquilibrée, comme indiqué ci-dessous.
• left_df: Quantité de données: Beaucoup, Exemple: Données réelles
• right_df: Quantité de données: petite, exemple: données de base

df = left_df.join(F.broadcast(right_df), on="id")

• union (): combine des trames de données verticalement

df = upper_df.union(bottom_df)

Opération de colonne (renommer, supprimer, sélectionner)

• withColumnRenamed (before, after): Renommer la colonne

Il est souvent utilisé lors de la lecture de csv sans nom de colonne.

#S'il n'y a pas de nom de colonne`_c0`De`_c{n}`Reçoit le nom de la colonne
df = df.withColumnRenamed("_c0", "id")

• select ("col_1", "col_2", ...): Obtenir par colonne

df = df.select("id")

• distinct (): suppression des doublons

df = df.select("id").distinct()
# count()Souvent utilisé en combinaison avec
#Exemple: numéro unique d'un certain identifiant
print(df.select("id").distinct().count())

• drop ("col_1", "col_2", ...): Drop par colonne

df = df.drop("id")

• drop Null Value

# simple
df = df.dropna()
# using subset
df = df.na.drop(subset=["lat", "lon"])

• collect_list (), collect_set (): Entrez les valeurs sous forme de liste dans une seule colonne

#Cas simple
df = df.select("id").select(F.collect_list("id"))
id_list = df.first()[0]

> id_list => ["A001", "A002", "B001"]

#Peut également être utilisé en combinaison avec groupBy
df = df.groupBy("id").agg(F.collect_set("code"), F.collect_list("name"))

> 
+---+-----------------+------------------+
| id|collect_set(code)|collect_list(name)|
+---+-----------------+------------------+
|  a|   [code1, code2]|           [name2]|
+---+-----------------+------------------+

• collect (): renvoie tous les éléments
• take (n): renvoie les n premiers éléments
• first (): renvoie le tout premier élément

#Obtenez directement la valeur de la trame de données
df = df.groupBy().avg()
avg_attribute = df.collect()[0]

> print(avg_attribute["avg({col_name})"])
{averaged_value}

filter

Vous pouvez utiliser F.col () pour appliquer le filtrage à des colonnes spécifiques

# using spark.function
df = df.filter(F.col("id") == "A001")

# pandas-like
df = df.filter(df['id'] == "A001")
df = df.filter(df.id == "A001")

• isin (): détermine si la valeur est dans un élément de la liste

Cependant, si possible, vous devez créer un cadre de données Spark à partir de date_list et le rejoindre.

df = df.filter(F.col("dt").isin(date_list))

orderBy

Le tri ne convient pas au traitement distribué, il vaut donc mieux ne pas en faire autant.

#Colonne unique uniquement
df = df.orderBy("count", ascending=False)

#Tri multi-conditions
df = df.orderBy(F.col("id").asc(), F.col("cound").desc())

groupBy (aggregate)

# count()
df = df.groupBy("id").count()

# multiple
# alias()Le nom de la colonne est modifié par la fonction
#Exemple: agrégation d'utilisateurs
df = df.groupBy("id").agg(
  F.count(F.lit(1)).alias("count"),
  F.mean(F.col("diff")).alias("diff_mean"),
  F.stddev(F.col("diff")).alias("diff_stddev"),
  F.min(F.col("diff")).alias("diff_min"),
  F.max(F.col("diff")).alias("diff_max")
)

> df.show()
(réduction)

# =======================
#Exemple: agrégation par date et heure de l'utilisateur
df = df.groupBy("id", "dt").agg(
  F.count(F.lit(1)).alias("count")
  )
  
> df.show()
+---+-----------+------+
| id|         dt| count|
+---+-----------+------+
|  a| 2020/01/01|     7|
|  a| 2020/01/02|     5|
|  a| 2020/01/03|     4|
+---+-----------+------+

# ===========================
#Exemple: agrégation par date / heure / emplacement utilisateur
df = df.groupBy("id", "dt", "location_id").agg(
  F.count(F.lit(1)).alias("count")
  )

> df.show()
+---+-----------+------------+------+
| id|         dt| location_id| count|
+---+-----------+------------+------+
|  a| 2020/01/01|           A|     2|
|  a| 2020/01/01|           B|     3|
|  a| 2020/01/01|           C|     2|
:   :           :            :      :
+---+-----------+------------+------+

• countDistinct (): supprimer les doublons lors de l'agrégation et du comptage

#Exemple: nombre d'utilisateurs uniques par date
df = df.groupBy("dt").agg(countDistinct("id").alias("id_count"))

> df.show()
+-----------+---------+
|         dt| id_count|
+-----------+---------+
| 2020/01/01|        7|
| 2020/01/02|        5|
| 2020/01/03|        4|
+-----------+---------+

# ===============================
#Exemple: nombre de jours pendant lesquels chaque utilisateur a été en contact au moins une fois
df = df.groupBy("id").agg(countDistinct("dt").alias("dt_count"))

> df.show()
+---+---------+
| id| dt_count|
+---+---------+
|  a|       10|
|  b|       15|
|  c|        4|
+---+---------+

• La liste peut également être spécifiée pour la spécification des éléments

group_columns = ["id", "dt"]
df = ad_touched_visit_df.groupBy(*group_columns).count()

fonction de fenêtre

• row_number (): Ajouter un numéro de ligne

w = Window().orderBy(F.col("id"))
df = df.withColumn("row_num", F.row_number().over(w))

• lag (): décalage des lignes

#Ajouter les données de la ligne précédente en tant que colonne
w = Window.partitionBy("id").orderBy("timestamp")
df = df.withColumn("prev_timestamp", F.lag(df["timestamp"]).over(w))

traitement en boucle

Il est fortement déconseillé car incompatible avec les environnements distribués. Il est préférable de ne l'utiliser que lorsque cela doit être fait.

for row in df.rdd.collect():
  do_some_with(row['id'])

Site de référence

• Opération de dataframe Pyspark
• Référence Spark SQL
• select_list