Support formation Microsoft Azure
DP-203
Analyser des données avec Apache Spark dans Azure Synapse Analytics
Gonzague Ducos
Introduction
Apache Spark est un framework de traitement parallèle open source pour le traitement et l’analytique à grande échelle des données.
Extrêmement populaire dans les scénarios de traitement de « Big Data »
Disponible dans plusieurs implémentations de plateforme, notamment :
Découvrir Apache Spark
Comment Spark fonctionne
Pools Spark dans Azure Synapse Analytics
Un cluster est implémenté en tant que pool Spark, qui fournit un runtime pour les opérations Spark. Création dans ou :
Azure Synapse Analytics sont serverless : ils démarrent à la demande et s’arrêtent lorsqu’ils sont inactifs.
Utiliser Spark dans Azure Synapse Analytics
Nombreux types d’application :
2 usages :
Bien qu’ils soient généralement utilisés de manière interactive, les notebooks peuvent être inclus dans des pipelines automatisés et s’exécuter en tant que script sans assistance.
Exécution de code Spark dans des notebooks
Accès aux données à partir d’un pool Synapse Spark
Utilisation la plus courante : utiliser des données dans un lac de données, où vous pouvez lire et écrire des fichiers dans plusieurs formats couramment utilisés, notamment du texte délimité, Parquet, Avro et autres
Analyser les données avec Spark
Exploration de données avec des dataframes
Structure de données appelée Jeu de données distribué résilient (RDD, resilient distributed dataset). Mais la structure de données la plus couramment utilisée pour utiliser des données structurées dans Spark est le dataframe.
Le code en Python :
%%pyspark
df = spark.read.load('abfss://container@store.dfs.core.windows.net/products.csv', format='csv', header=True)
display(df.limit(10))
La ligne %%pyspark au début est appelée magic.
Le même code en Scala :
%%spark
val df = spark.read.format("csv").option("header", "true").load("abfss://container@store.dfs.core.windows.net/products.csv")
display(df.limit(10))
Chargement des données dans un dataframe
%%pyspark
df = spark.read.load('abfss://container@store.dfs.core.windows.net/products.csv',
format='csv',
header=True
)
display(df.limit(10))
Spécification d’un schéma de dataframe
from pyspark.sql.types import *
from pyspark.sql.functions import *
productSchema = StructType([
StructField("ProductID", IntegerType()),
StructField("ProductName", StringType()),
StructField("Category", StringType()),
StructField("ListPrice", FloatType())
])
df = spark.read.load('abfss://container@store.dfs.core.windows.net/product-data.csv',
format='csv',
schema=productSchema,
header=False)
display(df.limit(10))
Filtrage et regroupement des dataframes
Utiliser les méthodes de la classe Dataframe pour filtrer, trier, regrouper et manipuler les données qu’elle contient :
pricelist_df = df.select("ProductID", "ListPrice")
Autre syntaxe de sélection de colonne :
pricelist_df = df["ProductID", "ListPrice"]
Chainer les méthodes :
bikes_df = df.select("ProductName", "ListPrice").where((df["Category"]=="Mountain Bikes") | (df["Category"]=="Road Bikes"))
display(bikes_df)
Regroupement :
counts_df = df.select("ProductID", "Category").groupBy("Category").count()
display(counts_df)
Utilisation d’expressions SQL dans Spark
L’API Dataframe fait partie d’une bibliothèque Spark appelée Spark SQL, qui permet aux analystes de données d’utiliser des expressions SQL pour interroger et manipuler des données
Création d’objets de base de données dans le catalogue Spark
Créer une vue temporaire est la méthode la plus simple pour rendre les données d’un dataframe disponibles pour pouvoir les interroger dans le catalogue Spark
df.createOrReplaceTempView("products")
spark.catalog.createTable : crée une table vide. Sont des structures de métadonnées qui stockent leurs données sous-jacentes dans l’emplacement de stockage associé au catalogue. La suppression d’une table supprime également ses données sous-jacentes.
saveAsTable : enregistrer un dataframe en tant que table.
spark.catalog.createExternalTable : crée une table externe.
Utilisation de l’API Spark SQL pour interroger des données
Interroge la table products en tant que dataframe.
bikes_df = spark.sql("SELECT ProductID, ProductName, ListPrice \
FROM products \
WHERE Category IN ('Mountain Bikes', 'Road Bikes')")
display(bikes_df)
Utilisation du code SQL
Dans un notebook, utiliser la commande magic %%sql pour exécuter le code SQL qui interroge les objets du catalogue :
%%sql
SELECT Category, COUNT(ProductID) AS ProductCount
FROM products
GROUP BY Category
ORDER BY Category
Want to print your doc?
This is not the way.
This is not the way.
Try clicking the ⋯ next to your doc name or using a keyboard shortcut (
CtrlP
) instead.