Oktatóanyag: Adatok betöltése és átalakítása Az Apache Spark DataFrame-ek használatával

Ez az oktatóanyag bemutatja, hogyan tölthet be és alakíthat át adatokat az Apache Spark Python (PySpark) DataFrame API, az Apache Spark Scala DataFrame API és a SparkR SparkDataFrame API használatával az Azure Databricksben.

Az oktatóanyag végére megismerheti a DataFrame-et, és megismerkedhet a következő feladatokkal:

Python

• Változók definiálása és nyilvános adatok másolása Unity Catalog-kötetbe
• DataFrame létrehozása Pythonnal
• Adatok betöltése DataFrame-fájlba CSV-fájlból
• DataFrame megtekintése és használata
• A DataFrame mentése
• SQL-lekérdezések futtatása a PySparkban

Lásd még az Apache Spark PySpark API-referenciát.

Scala

• Változók definiálása és nyilvános adatok másolása Unity Catalog-kötetbe
• DataFrame létrehozása a Scalával
• Adatok betöltése DataFrame-fájlba CSV-fájlból
• DataFrame megtekintése és használata
• A DataFrame mentése
• SQL-lekérdezések futtatása az Apache Sparkban

Lásd még az Apache Spark Scala API-referenciát.

R

• Változók definiálása és nyilvános adatok másolása Unity Catalog-kötetbe
• SparkR SparkDataFrame-ek létrehozása
• Adatok betöltése DataFrame-fájlba CSV-fájlból
• DataFrame megtekintése és használata
• A DataFrame mentése
• SQL-lekérdezések futtatása a SparkR-ben

Lásd még : Apache SparkR API-referencia.

Mi az a DataFrame?

A DataFrame egy kétdimenziós címkézett adatstruktúra, amely különböző típusú oszlopokat tartalmaz. A DataFrame-ekre, például számolótáblákra, SQL-táblákra vagy sorozatobjektumok szótárára is gondolhat. Az Apache Spark DataFrame-ek számos függvényt biztosítanak (oszlopok kiválasztása, szűrés, illesztés, összesítés), amelyek lehetővé teszik a gyakori adatelemzési problémák hatékony megoldását.

Az Apache Spark DataFrame-ek rugalmas elosztott adathalmazokra (RDD-kre) épülő absztrakciók. A Spark DataFrames és a Spark SQL egységes tervezési és optimalizálási motort használ, amely lehetővé teszi, hogy az Azure Databricks (Python, SQL, Scala és R) összes támogatott nyelvén szinte azonos teljesítményt nyújtsunk.

Követelmények

A következő oktatóanyag elvégzéséhez meg kell felelnie a következő követelményeknek:

• Az oktatóanyagban szereplő példák használatához a munkaterületen engedélyezve kell legyen a Unity Catalogus.

• Az oktatóanyagban szereplő példák egy Unity-katalógust kötetet a mintaadatok tárolására. A példák használatához hozzon létre egy kötetet, és használja a kötet katalógusát, sémáját és kötetneveit a példák által használt kötetútvonal beállításához.

• A Unity Katalógusban a következő engedélyekkel kell rendelkeznie:

  • READ VOLUMEvagy WRITE VOLUMEALL PRIVILEGES az oktatóanyaghoz használt kötethez.
  • USE SCHEMA vagy ALL PRIVILEGES az oktatóanyaghoz használt sémához.
  • USE CATALOG vagy ALL PRIVILEGES az oktatóanyaghoz használt katalógushoz.

  Az engedélyek beállításához tekintse meg a Databricks-rendszergazdai vagy a Unity Catalog-jogosultságokat és a biztonságos objektumokat.

Tipp.

A cikkhez tartozó kész jegyzetfüzetekért tekintse meg a DataFrame oktatóanyag-jegyzetfüzeteit.

1. lépés: Változók definiálása és CSV-fájl betöltése

Ez a lépés meghatározza az oktatóanyagban használt változókat, majd betölt egy CSV-fájlt, amely a health.data.ny.gov babanévadatait tartalmazza a Unity Catalog-kötetbe.

1. Nyisson meg egy új jegyzetfüzetet az ikonra kattintva. Az Azure Databricks-jegyzetfüzetek közötti navigálásról a A jegyzetfüzet megjelenésének testreszabásacímű témakörben olvashat.

2. Másolja és illessze be a következő kódot az új üres jegyzetfüzetcellába. Cserélje le a <catalog-name>, <schema-name>és <volume-name> elemeket a Unity Catalog kötetének katalógus-, séma- és kötetneveire. Cserélje le <table_name> egy tetszőleges táblanévre. Az oktatóanyag későbbi részében babanévadatokat fog betölteni ebbe a táblába.

   Python

   catalog = "<catalog_name>"
   schema = "<schema_name>"
   volume = "<volume_name>"
   download_url = "https://health.data.ny.gov/api/views/jxy9-yhdk/rows.csv"
   file_name = "rows.csv"
   table_name = "<table_name>"
   path_volume = "/Volumes/" + catalog + "/" + schema + "/" + volume
   path_table = catalog + "." + schema
   print(path_table) # Show the complete path
   print(path_volume) # Show the complete path
   

   Scala

   val catalog = "<catalog_name>"
   val schema = "<schema_name>"
   val volume = "<volume_name>"
   val downloadUrl = "https://health.data.ny.gov/api/views/jxy9-yhdk/rows.csv"
   val fileName = "rows.csv"
   val tableName = "<table_name>"
   val pathVolume = s"/Volumes/$catalog/$schema/$volume"
   val pathTable = s"$catalog.$schema"
   print(pathVolume) // Show the complete path
   print(pathTable) // Show the complete path
   

   R

   catalog <- "<catalog_name>"
   schema <- "<schema_name>"
   volume <- "<volume_name>"
   download_url <- "https://health.data.ny.gov/api/views/jxy9-yhdk/rows.csv"
   file_name <- "rows.csv"
   table_name <- "<table_name>"
   path_volume <- paste("/Volumes/", catalog, "/", schema, "/", volume, sep = "")
   path_table <- paste(catalog, ".", schema, sep = "")
   print(path_volume) # Show the complete path
   print(path_table) # Show the complete path
   

3. Nyomja le Shift+Enter a cellát, és hozzon létre egy új üres cellát.

4. Másolja és illessze be a következő kódot az új üres jegyzetfüzetcellába. Ez a kód a rows.csv fájlt a health.data.ny.gov címről a Unity Catalog kötetébe másolja a Databricks dbutils paranccsal.

   Python

   dbutils.fs.cp(f"{download_url}", f"{path_volume}/{file_name}")
   

   Scala

   dbutils.fs.cp(downloadUrl, s"$pathVolume/$fileName")
   

   R

   dbutils.fs.cp(download_url, paste(path_volume, "/", file_name, sep = ""))
   

5. Nyomja le Shift+Enter a cellát, majd lépjen a következő cellára.

2. lépés: DataFrame létrehozása

Ez a lépés létrehoz egy tesztadatokkal elnevezett df1 DataFrame-et, majd megjeleníti annak tartalmát.

1. Másolja és illessze be a következő kódot az új üres jegyzetfüzetcellába. Ez a kód létrehozza a DataFrame-et tesztadatokkal, majd megjeleníti a DataFrame tartalmát és sémáját.

   Python

   data = [[2021, "test", "Albany", "M", 42]]
   columns = ["Year", "First_Name", "County", "Sex", "Count"]
   
   # highlight-next-line
   df1 = spark.createDataFrame(data, schema="Year int, First_Name STRING, County STRING, Sex STRING, Count int")
   display(df1) # The display() method is specific to Databricks notebooks and provides a richer visualization.
   # df1.show() The show() method is a part of the Apache Spark DataFrame API and provides basic visualization.
   

   Scala

   val data = Seq((2021, "test", "Albany", "M", 42))
   val columns = Seq("Year", "First_Name", "County", "Sex", "Count")
   
   // highlight-next-line
   val df1 = data.toDF(columns: _*)
   display(df1) // The display() method is specific to Databricks notebooks and provides a richer visualization.
   // df1.show() The show() method is a part of the Apache Spark DataFrame API and provides basic visualization.
   

   R

   # Load the SparkR package that is already preinstalled on the cluster.
   library(SparkR)
   
   data <- data.frame(
     Year = as.integer(c(2021)),
     First_Name = c("test"),
     County = c("Albany"),
     Sex = c("M"),
     Count = as.integer(c(42))
   )
   
   # highlight-next-line
   df1 <- createDataFrame(data)
   display(df1) # The display() method is specific to Databricks notebooks and provides a richer visualization.
   # head(df1) The head() method is a part of the Apache SparkR DataFrame API and provides basic visualization.
   

2. Nyomja le Shift+Enter a cellát, majd lépjen a következő cellára.

3. lépés: Adatok betöltése adatkeretbe CSV-fájlból

Ez a lépés létrehoz egy df_csv nevű DataFrame-et a korábban az Ön Unity Catalog-kötetébe betöltött CSV-fájlból. Lásd: spark.read.csv.

1. Másolja és illessze be a következő kódot az új üres jegyzetfüzetcellába. Ez a kód betölti a babanév adatait a DataFrame-be df_csv a CSV-fájlból, majd megjeleníti a DataFrame tartalmát.

   Python

   df_csv = spark.read.csv(f"{path_volume}/{file_name}",
       header=True,
       inferSchema=True,
       sep=",")
   display(df_csv)
   

   Scala

   val dfCsv = spark.read
       .option("header", "true")
       .option("inferSchema", "true")
       .option("delimiter", ",")
       .csv(s"$pathVolume/$fileName")
   
   display(dfCsv)
   

   R

   df_csv <- read.df(paste(path_volume, "/", file_name, sep=""),
       source="csv",
       header = TRUE,
       inferSchema = TRUE,
       delimiter = ",")
   
   display(df_csv)
   

2. Nyomja le Shift+Enter a cellát, majd lépjen a következő cellára.

Számos támogatott fájlformátumból tölthet be adatokat.

4. lépés: A DataFrame megtekintése és használata

Tekintse meg és használja a DataFrames babaneveit az alábbi módszerekkel.

A DataFrame-séma nyomtatása

Megtudhatja, hogyan jelenítheti meg az Apache Spark DataFrame sémáját. Az Apache Spark a séma kifejezést használja a DataFrame oszlopainak nevére és adattípusára való hivatkozáshoz.

Feljegyzés

Az Azure Databricks a kifejezésséma használatával is leírja a katalógusban regisztrált táblák gyűjteményét.

1. Másolja és illessze be a következő kódot egy üres jegyzetfüzetcellába. Ez a kód a DataFrame-ek sémáját mutatja be a .printSchema() metódussal a két DataFrame sémáinak megtekintéséhez – a két DataFrame egyesítésére való felkészüléshez.

   Python

   df_csv.printSchema()
   df1.printSchema()
   

   Scala

   dfCsv.printSchema()
   df1.printSchema()
   

   R

   printSchema(df_csv)
   printSchema(df1)
   

2. Nyomja le Shift+Enter a cellát, majd lépjen a következő cellára.

Adatkeret oszlopának átnevezése

Megtudhatja, hogyan nevezhet át egy oszlopot egy DataFrame-ben.

1. Másolja és illessze be a következő kódot egy üres jegyzetfüzetcellába. Ez a kód átnevez egy oszlopot a df1_csv DataFrame-ben, hogy megfeleljen a df1 DataFrame megfelelő oszlopának. Ez a kód az Apache Spark withColumnRenamed() metódust használja.

   Python

   df_csv = df_csv.withColumnRenamed("First Name", "First_Name")
   df_csv.printSchema
   

   Scala

   val dfCsvRenamed = dfCsv.withColumnRenamed("First Name", "First_Name")
   // when modifying a DataFrame in Scala, you must assign it to a new variable
   dfCsvRenamed.printSchema()
   

   R

   df_csv <- withColumnRenamed(df_csv, "First Name", "First_Name")
   printSchema(df_csv)
   

2. Nyomja le Shift+Enter a cellát, majd lépjen a következő cellára.

Adatkeretek egyesítése

Megtudhatja, hogyan hozhat létre egy új DataFrame-et, amely hozzáadja az egyik DataFrame sorait a másikhoz.

1. Másolja és illessze be a következő kódot egy üres jegyzetfüzetcellába. Ez a kód az Apache Spark union() metódussal egyesíti az első DataFrame df tartalmát a CSV-fájlból betöltött babaneveket tartalmazó DataFrame-lel df_csv .

   Python

   df = df1.union(df_csv)
   display(df)
   

   Scala

   val df = df1.union(dfCsvRenamed)
   display(df)
   

   R

   display(df <- union(df1, df_csv))
   

2. Nyomja le Shift+Enter a cellát, majd lépjen a következő cellára.

Sorok szűrése DataFrame-ben

Az adatkészlet legnépszerűbb babaneveit sorok szűrésével fedezheti fel az Apache Spark .filter() vagy .where() metódusok használatával. Szűréssel kiválaszthatja a dataframe-ben visszaadni vagy módosítani kívánt sorok egy részhalmazát. Nincs különbség a teljesítményben vagy a szintaxisban, ahogy az alábbi példákban látható.

.filter() metódus használata

1. Másolja és illessze be a következő kódot egy üres jegyzetfüzetcellába. Ez a kód az Apache Spark .filter() metódussal jeleníti meg azokat a sorokat a DataFrame-ben, amelyek száma meghaladja az 50-et.

   Python

   display(df.filter(df["Count"] > 50))
   

   Scala

   display(df.filter(df("Count") > 50))
   

   R

   display(filteredDF <- filter(df, df$Count > 50))
   

2. Nyomja le Shift+Enter a cellát, majd lépjen a következő cellára.

.where() metódus használata

1. Másolja és illessze be a következő kódot egy üres jegyzetfüzetcellába. Ez a kód az Apache Spark .where() metódussal jeleníti meg azokat a sorokat a DataFrame-ben, amelyek száma meghaladja az 50-et.

   Python

   display(df.where(df["Count"] > 50))
   

   Scala

   display(df.where(df("Count") > 50))
   

   R

   display(filtered_df <- where(df, df$Count > 50))
   

2. Nyomja le Shift+Enter a cellát, majd lépjen a következő cellára.

Oszlopok kijelölése DataFrame-ből és sorrend gyakoriság szerint

Ismerje meg, hogyan adhatja meg a visszaadni kívánt DataFrame oszlopait a select() módszerrel, hogy megtudja a babanév gyakoriságát. Az eredményeket az Apache Spark orderby és desc a függvények segítségével rendezheti.

Az Apache Spark pyspark.sql modulja támogatja az SQL-függvényeket. Az oktatóanyagban használt függvények közé tartozik az Apache Spark orderBy(), desc(), és expr() függvény. Ezeknek a függvényeknek a használatát úgy engedélyezheti, hogy szükség szerint importálja őket a munkamenetbe.

1. Másolja és illessze be a következő kódot egy üres jegyzetfüzetcellába. Ez a kód importálja a függvényt desc() , majd az Apache Spark select() metódust és az Apache Sparkot orderBy() és desc() függvényeket használja a leggyakoribb nevek és azok számának csökkenő sorrendben való megjelenítéséhez.

   Python

   from pyspark.sql.functions import desc
   display(df.select("First_Name", "Count").orderBy(desc("Count")))
   

   Scala

   import org.apache.spark.sql.functions.desc
   display(df.select("First_Name", "Count").orderBy(desc("Count")))
   

   R

   display(arrange(select(df, df$First_Name, df$Count), desc(df$Count)))
   

2. Nyomja le Shift+Enter a cellát, majd lépjen a következő cellára.

Adatkeret-részhalmaz létrehozása

Megtudhatja, hogyan hozhat létre adatkeret-részhalmazt egy meglévő DataFrame-ből.

1. Másolja és illessze be a következő kódot egy üres jegyzetfüzetcellába. Ez a kód az Apache Spark filter metódus használatával hoz létre egy új DataFrame-et, amely év, darabszám és nem szerint korlátozza az adatokat. Az Apache Spark select() metódust használja az oszlopok korlátozására. Emellett az Apache Spark orderBy() és desc() a függvények használatával rendezi az új DataFrame-et szám szerint.

   Python

   subsetDF = df.filter((df["Year"] == 2009) & (df["Count"] > 100) & (df["Sex"] == "F")).select("First_Name", "County", "Count").orderBy(desc("Count"))
   display(subsetDF)
   

   Scala

   val subsetDF = df.filter((df("Year") === 2009) && (df("Count") > 100) && (df("Sex") === "F")).select("First_Name", "County", "Count").orderBy(desc("Count"))
   
   display(subsetDF)
   

   R

   subsetDF <- select(filter(df, (df$Count > 100) & (df$year == 2009) & df["Sex"] == "F")), "First_Name", "County", "Count")
   display(subsetDF)
   

2. Nyomja le Shift+Enter a cellát, majd lépjen a következő cellára.

5. lépés: A DataFrame mentése

Megtudhatja, hogyan menthet DataFrame-eket. A DataFrame-et mentheti egy táblába, vagy fájlba vagy több fájlba is írhatja a DataFrame-et.

A DataFrame mentése táblázatba

Az Azure Databricks alapértelmezés szerint az összes tábla Delta Lake-formátumát használja. A DataFrame mentéséhez CREATE táblajogokkal kell rendelkeznie a katalógusban és a sémában.

1. Másolja és illessze be a következő kódot egy üres jegyzetfüzetcellába. Ez a kód az oktatóanyag elején definiált változó használatával menti a DataFrame tartalmát egy táblába.

   Python

   df.write.mode("overwrite").saveAsTable(f"{path_table}.{table_name}")
   

   Scala

   df.write.mode("overwrite").saveAsTable(s"$pathTable" + "." + s"$tableName")
   

   R

   saveAsTable(df, paste(path_table, ".", table_name), mode = "overwrite")
   

2. Nyomja le Shift+Enter a cellát, majd lépjen a következő cellára.

A legtöbb Apache Spark-alkalmazás nagy adatkészleteken és elosztott módon működik. Az Apache Spark egyetlen fájl helyett fájlkönyvtárat ír ki. A Delta Lake felosztja a Parquet-mappákat és -fájlokat. Sok adatrendszer képes beolvasni ezeket a fájlkönyvtárakat. Az Azure Databricks azt javasolja, hogy a legtöbb alkalmazáshoz használja a táblákat fájlelérési utakon keresztül.

A DataFrame mentése JSON-fájlokba

1. Másolja és illessze be a következő kódot egy üres jegyzetfüzetcellába. Ez a kód jSON-fájlok könyvtárába menti a DataFrame-et.

   Python

   df.write.format("json").mode("overwrite").save("/tmp/json_data")
   

   Scala

   df.write.format("json").mode("overwrite").save("/tmp/json_data")
   

   R

   write.df(df, path = "/tmp/json_data", source = "json", mode = "overwrite")
   

2. Nyomja le Shift+Enter a cellát, majd lépjen a következő cellára.

A DataFrame beolvasása egy JSON-fájlból

Megtudhatja, hogyan olvashatja be a JSON-adatokat egy könyvtárból egy DataFrame-be az Apache Spark spark.read.format() metódus használatával.

1. Másolja és illessze be a következő kódot egy üres jegyzetfüzetcellába. Ez a kód megjeleníti az előző példában mentett JSON-fájlokat.

   Python

   display(spark.read.format("json").json("/tmp/json_data"))
   

   Scala

   display(spark.read.format("json").json("/tmp/json_data"))
   

   R

   display(read.json("/tmp/json_data"))
   

2. Nyomja le Shift+Enter a cellát, majd lépjen a következő cellára.

További feladatok: SQL-lekérdezések futtatása a PySparkban, a Scalában és az R-ben

Az Apache Spark DataFrame-ek az alábbi lehetőségeket biztosítják az SQL és a PySpark, a Scala és az R kombinálásához. Az alábbi kódot az oktatóanyaghoz létrehozott jegyzetfüzetben futtathatja.

Adjon meg egy oszlopot SQL-lekérdezésként

Ismerje meg, hogyan használhatja az Apache Spark selectExpr() metódust. Ez a metódus egy változata, amely elfogadja az select() SQL-kifejezéseket, és egy frissített DataFrame-et ad vissza. Ez a módszer lehetővé teszi egy SQL-kifejezés, például uppera .

1. Másolja és illessze be a következő kódot egy üres jegyzetfüzetcellába. Ez a kód az Apache Spark selectExpr() metódust és az SQL upper kifejezést használja a sztringoszlop nagybetűssé alakításához (és az oszlop átnevezéséhez).

   Python

   display(df.selectExpr("Count", "upper(County) as big_name"))
   

   Scala

   display(df.selectExpr("Count", "upper(County) as big_name"))
   

   R

   display(df_selected <- selectExpr(df, "Count", "upper(County) as big_name"))
   

2. Nyomja le Shift+Enter a cellát, majd lépjen a következő cellára.

A expr() használata a SQL-szintaxis alkalmazásához egy oszlophoz

Megtudhatja, hogyan importálhatja és használhatja az Apache Spark expr() függvényt az SQL-szintaxis használatára bárhol, ahol egy oszlop meg lesz adva.

1. Másolja és illessze be a következő kódot egy üres jegyzetfüzetcellába. Ez a kód importálja a expr() függvényt, majd az Apache Spark expr() függvényt és az SQL lower kifejezést használja egy sztringoszlop kisbetűssé alakításához (és átnevezi az oszlopot).

   Python

   from pyspark.sql.functions import expr
   display(df.select("Count", expr("lower(County) as little_name")))
   

   Scala

   import org.apache.spark.sql.functions.{col, expr}
   // Scala requires us to import the col() function as well as the expr() function
   
   display(df.select(col("Count"), expr("lower(County) as little_name")))
   

   R

   display(df_selected <- selectExpr(df, "Count", "lower(County) as little_name"))
   # expr() function is not supported in R, selectExpr in SparkR replicates this functionality
   

2. Nyomja le Shift+Enter a cellát, majd lépjen a következő cellára.

Tetszőleges SQL-lekérdezés futtatása spark.sql() függvénnyel

Ismerje meg, hogyan futtathat tetszőleges SQL-lekérdezéseket az Apache Spark spark.sql() függvény használatával.

1. Másolja és illessze be a következő kódot egy üres jegyzetfüzetcellába. Ez a kód az Apache Spark spark.sql() függvénnyel kérdez le egy SQL-táblát SQL-szintaxis használatával.

   Python

   display(spark.sql(f"SELECT * FROM {path_table}.{table_name}"))
   

   Scala

   display(spark.sql(s"SELECT * FROM $pathTable.$tableName"))
   

   R

   display(sql(paste("SELECT * FROM", path_table, ".", table_name)))
   

2. Nyomja le Shift+Enter a cellát, majd lépjen a következő cellára.

DataFrame-oktatóanyag-jegyzetfüzetek

Az alábbi jegyzetfüzetek az oktatóanyagból származó példákat tartalmazzák.

Python

DataFrames-oktatóanyag a Python használatával

Jegyzetfüzet szerezz

Scala

DataFrames-oktatóanyag a Scala használatával

Jegyzetfüzet szerezz

R

DataFrames-oktatóanyag az R használatával

Jegyzetfüzet szerezz

További erőforrások

• PySpark az Azure Databricksben
• Apache Spark API-k referenciája
• Konvertálás a PySpark és a pandas DataFrames között
• Pandas API a Sparkban