แชร์ผ่าน

ใช้ sparklyr

  • บทความ

Sparklyr คืออินเทอร์เฟซ R สําหรับ Apache Spark ซึ่งมีกลไกในการโต้ตอบกับ Spark โดยใช้อินเทอร์เฟซ R ที่คุ้นเคย คุณสามารถใช้ sparklyr ผ่านข้อกําหนดงานแบบชุดงานของ Spark หรือด้วยสมุดบันทึก Microsoft Fabric แบบโต้ตอบ

sparklyrใช้พร้อมกับแพคเกจเรียบร้อยอื่น ๆ เช่น dplyr Microsoft Fabric กระจาย sparklyr รุ่นเสถียรล่าสุดและเป็นระเบียบกับทุกการเผยแพร่รันไทม์ คุณสามารถนําเข้าและเริ่มต้นใช้ API

ข้อกำหนดเบื้องต้น

  • เปิดหรือสร้างสมุดบันทึก หากต้องการเรียนรู้วิธีการ ดู วิธีใช้สมุดบันทึก Microsoft Fabric

  • ตั้งค่าตัวเลือกภาษาเป็น SparkR (R) เพื่อเปลี่ยนภาษาหลัก

  • แนบสมุดบันทึกของคุณเข้ากับเลคเฮ้าส์ ทางด้านซ้าย เลือก เพิ่ม เพื่อเพิ่มเลคเฮาส์ที่มีอยู่ หรือเพื่อสร้างเลคเฮ้าส์

เชื่อมต่อ sparklyr ไปยังคลัสเตอร์ Synapse Spark

ใช้วิธีการเชื่อมต่อต่อไปนี้ใน spark_connect() เพื่อสร้าง sparklyr การเชื่อมต่อ เราสนับสนุนวิธีการเชื่อมต่อใหม่ที่เรียกว่า synapseซึ่งช่วยให้คุณสามารถเชื่อมต่อกับเซสชัน Spark ที่มีอยู่ได้ ซึ่งจะช่วยลดเวลาเริ่มต้นเซสชัน sparklyr ได้อย่างมาก นอกจากนี้เรายังร่วมให้วิธีการเชื่อมต่อนี้กับโครงการ sparklyr โอเพนซอร์ส (Open Source)d ด้วย method = "synapse"คุณสามารถใช้ทั้ง sparklyr และ SparkR ในเซสชันเดียวกัน และแชร์ข้อมูลระหว่างกันได้อย่างง่ายดาย

# connect sparklyr to your spark cluster
spark_version <- sparkR.version()
config <- spark_config()
sc <- spark_connect(master = "yarn", version = spark_version, spark_home = "/opt/spark", method = "synapse", config = config)

ใช้ sparklyr เพื่ออ่านข้อมูล

เซสชัน Spark ใหม่ไม่มีข้อมูล ขั้นตอนแรกคือการโหลดข้อมูลลงในหน่วยความจําของเซสชัน Spark หรือชี้ Spark ไปยังตําแหน่งที่ตั้งของข้อมูลเพื่อให้สามารถเข้าถึงข้อมูลได้ตามต้องการ

# load the sparklyr package
library(sparklyr)

# copy data from R environment to the Spark session's memory
mtcars_tbl <- copy_to(sc, mtcars, "spark_mtcars", overwrite = TRUE)

head(mtcars_tbl)

การใช้ sparklyrคุณยังสามารถ write และข้อมูล read จากไฟล์ Lakehouse โดยใช้เส้นทาง ABFS ได้ หากต้องการอ่านและเขียนไปยังเลคเฮ้าส์ ก่อนอื่นให้เพิ่มชุดข้อมูลลงในเซสชั่นของคุณ ทางด้านซ้ายของสมุดบันทึก ให้เลือก เพิ่ม เพื่อเพิ่มเลคเฮ้าส์ที่มีอยู่แล้ว หรือสร้างเลคเฮ้าส์

หากต้องการค้นหาเส้นทาง ABFS ของคุณ ให้คลิกขวาบน โฟลเดอร์ไฟล์ ใน Lakehouse ของคุณจากนั้นเลือก คัดลอกเส้นทาง ABFS วางเส้นทางของคุณเพื่อแทนที่ abfss://xxxx@onelake.dfs.fabric.microsoft.com/xxxx/Files ในโค้ดนี้:

temp_csv = "abfss://xxxx@onelake.dfs.fabric.microsoft.com/xxxx/Files/data/mtcars.csv"

# write the table to your lakehouse using the ABFS path
spark_write_csv(mtcars_tbl, temp_csv, header = TRUE, mode = 'overwrite')

# read the data as CSV from lakehouse using the ABFS path
mtcarsDF <- spark_read_csv(sc, temp_csv) 
head(mtcarsDF)

ใช้ sparklyr เพื่อจัดการข้อมูล

sparklyr จัดเตรียมหลายวิธีในการประมวลผลข้อมูลภายใน Spark โดยใช้:

  • dplyr คำ สั่ง
  • SparkSQL
  • ตัวแปลงคุณลักษณะของ Spark

ใช้ dplyr

คุณสามารถใช้คําสั่งที่คุ้นเคย dplyr เพื่อเตรียมข้อมูลภายใน Spark ได้ คําสั่งทํางานภายใน Spark ดังนั้นจึงไม่มีการส่งข้อมูลที่ไม่จําเป็นระหว่าง R และ Spark

คลิกที่จัดการข้อมูลด้วยdplyrเพื่อดูเอกสารเพิ่มเติมเกี่ยวกับการใช้ dplyr กับ Spark

# count cars by the number of cylinders the engine contains (cyl), order the results descendingly
library(dplyr)

cargroup <- group_by(mtcars_tbl, cyl) %>%
  count() %>%
  arrange(desc(n))

cargroup

sparklyr และ dplyr แปลคําสั่ง R ลงใน Spark SQL สําหรับเรา เมื่อต้องการดูผลลัพธ์คิวรี ให้ใช้ show_query():

# show the dplyr commands that are to run against the Spark connection
dplyr::show_query(cargroup)

ใช้ SQL

นอกจากนี้ยังเป็นไปได้ที่จะดําเนินการคิวรี SQL โดยตรงกับตารางภายในคลัสเตอร์ Spark วัตถุ spark_connection() ใช้ อินเทอร์เฟซ DBI สําหรับ Spark ดังนั้นคุณสามารถใช้ dbGetQuery() เพื่อดําเนินการ SQL และส่งกลับผลลัพธ์เป็นเฟรมข้อมูล R:

library(DBI)
dbGetQuery(sc, "select cyl, count(*) as n from spark_mtcars
GROUP BY cyl
ORDER BY n DESC")

ใช้ตัวแปลงคุณลักษณะ

ทั้งสองวิธีก่อนหน้านี้ขึ้นอยู่กับคําสั่ง SQL Spark มีคําสั่งที่ทําให้การแปลงข้อมูลบางอย่างสะดวกยิ่งขึ้นและไม่มีการใช้ SQL

ตัวอย่างเช่น ft_binarizer() คําสั่งลดความซับซ้อนของการสร้างคอลัมน์ใหม่ที่ระบุว่าค่าของคอลัมน์อื่นอยู่เหนือค่าเกณฑ์บางอย่างหรือไม่

คุณสามารถค้นหารายการทั้งหมดของตัวแปลงคุณลักษณะ Spark ที่พร้อมใช้งานผ่าน sparklyr การอ้างอิง -FT

mtcars_tbl %>% 
  ft_binarizer("mpg", "over_20", threshold = 20) %>% 
  select(mpg, over_20) %>% 
  head(5)

แชร์ข้อมูลระหว่าง sparklyr และ SparkR

เมื่อคุณ เชื่อมต่อกับ sparklyr synapse spark cluster ด้วย method = "synapse"คุณสามารถใช้ทั้ง sparklyr และ SparkR ในเซสชันเดียวกัน และแชร์ข้อมูลระหว่างกันได้อย่างง่ายดาย คุณสามารถสร้างตาราง Spark ใน sparklyr และอ่านจากSparkR

# load the sparklyr package
library(sparklyr)

# Create table in `sparklyr`
mtcars_sparklyr <- copy_to(sc, df = mtcars, name = "mtcars_tbl", overwrite = TRUE, repartition = 3L)

# Read table from `SparkR`
mtcars_sparklr <- SparkR::sql("select cyl, count(*) as n
from mtcars_tbl
GROUP BY cyl
ORDER BY n DESC")

head(mtcars_sparklr)

การเรียนรู้ของเครื่อง

นี่คือตัวอย่างที่เราใช้ ml_linear_regression() เพื่อให้พอดีกับแบบจําลองการถดถอยเชิงเส้น เราใช้ชุดข้อมูลในตัว mtcars และดูว่าเราสามารถคาดการณ์ปริมาณการใช้น้ํามันเชื้อเพลิงของรถยนต์ (mpg) ตามน้ําหนัก (wt) และจํานวนถังที่เครื่องยนต์ประกอบด้วย (cyl) ได้หรือไม่ เราคาดเดาในแต่ละกรณีที่ความสัมพันธ์ระหว่าง mpg และแต่ละคุณลักษณะของเราเป็นแบบเชิงเส้น

สร้างชุดข้อมูลการทดสอบและการฝึกอบรม

ใช้การแยก 70% สําหรับการฝึกและ 30% สําหรับการทดสอบแบบจําลอง การเล่นด้วยอัตราส่วนนี้ส่งผลให้เกิดแบบจําลองที่แตกต่างกัน

# split the dataframe into test and training dataframes

partitions <- mtcars_tbl %>%
  select(mpg, wt, cyl) %>% 
  sdf_random_split(training = 0.7, test = 0.3, seed = 2023)

ฝึกแบบจำลอง

ฝึกแบบจําลองการถดถอยโลจิสติกส์

fit <- partitions$training %>%
  ml_linear_regression(mpg ~ .)

fit

ตอนนี้ใช้ summary() เพื่อเรียนรู้เพิ่มเติมเล็กน้อยเกี่ยวกับคุณภาพของแบบจําลองของเราและความสําคัญทางสถิติของแต่ละตัวคาดการณ์ของเรา

summary(fit)

ใช้แบบจําลอง

คุณสามารถใช้แบบจําลองบนชุดข้อมูลทดสอบโดย ml_predict()การเรียก

pred <- ml_predict(fit, partitions$test)

head(pred)

สําหรับรายการของแบบจําลอง Spark ML ที่พร้อมใช้งานผ่าน sparklyr เยี่ยมชม การอ้างอิง - ML

ยกเลิกการเชื่อมต่อจากคลัสเตอร์ Spark

คุณสามารถเรียกใช้ spark_disconnect() หรือเลือก ปุ่ม หยุดเซสชัน ที่ด้านบนของริบบอนสมุดบันทึกจะสิ้นสุดเซสชัน Spark ของคุณ

spark_disconnect(sc)

เนื้อหาที่เกี่ยวข้อง

เรียนรู้เพิ่มเติมเกี่ยวกับฟังก์ชันการทํางานของ R: