Làm việc với Spark DataFrames – Truy vấn cơ bản

Spark DataFrames

DataFrame là một kiểu dữ liệu collection phân tán, được tổ chức thành các cột được đặt tên. Về mặt khái niệm, nó tương đương với các bảng quan hệ (relational tables) đi kèm với các kỹ thuật tối ưu tính toán.

DataFrame có thể được xây dựng từ nhiều nguồn dữ liệu khác nhau như Hive table, các file dữ liệu có cấu trúc hay bán cấu trúc (csv, json), các hệ cơ sở dữ liệu phổ biến (MySQL, MongoDB, Cassandra), hoặc RDDs hiện hành. API này được thiết kế cho các ứng dụng Big Data và Data Science hiện đại. Kiểu dữ liệu này được lấy cảm hứng từ DataFrame trong Lập trình R và Pandas trong Python hứa hẹn mang lại hiệu suất tính toán cao hơn.

Spark DataFrames Performance

Trong bài viết này, tôi sẽ tiến hành thực nghiệm một vài truy vấn cơ bản để làm quen với kiểu dữ liệu DataFrames. Ở các bài viết sau, ta sẽ đi sâu hơn vào các truy vấn nâng cao và phức tạp hơn.

Chuẩn bị

Cài đặt môi trường lập trình

Bạn có thể sử dụng một trong cách cách sau để deploy một ứng dụng Spark

• Command line
• SBT
• IntelliJ IDEA

Cơ sở dữ liệu

Để đơn giản, tôi sẽ lấy dữ liệu từ www.w3schools.com và tổng hợp thành các file *.csv để dễ dàng nạp dữ liệu đầu vào cho chương trình minh hoạ. Bạn có thể download trực tiếp các file này ở đây.

Thiết lập cho file build.sbt

Ở đây, tôi sẽ quản lý các dependencies thông qua sbt. Bên dưới là các dependencies các bạn cần để tiến hành các thực nghiệm truy vấn dữ liệu.

name := "spark-dataframes"

version := "1.0"

scalaVersion := "2.10.5"

libraryDependencies ++= Seq(
  "org.apache.spark" % "spark-core_2.10" % "1.6.1",
  "org.apache.spark" % "spark-sql_2.10" % "1.6.1",
  "com.databricks" % "spark-csv_2.10" % "1.4.0"
)

Chạy một truy vấn đơn giản

Đây là một đoạn code scala đơn giản. Công việc chính là đọc vào file customer.csv sau đó lưu vào DataFrames df để hiển thị. Hàm registerTempTable(“Customers”) dùng để định danh cho sqlContext biết tên của bảng dữ liệu bạn đang làm việc. Toàn bộ code, bạn có thể sao chép từ Github.

import org.apache.spark._
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext

object MainExecutor {
  def main(args: Array[String]) {

    val conf = new SparkConf()
      .setAppName("Spark DataFrames Application")
      .setMaster("local[2]")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)

    // Loading customers data
    val df = sqlContext.read
      .format("com.databricks.spark.csv")
      .option("header", "true") // Use first line of all files as header
      .option("inferSchema", "true") // Automatically infer data types
      .load("db/csv/customers.csv")

    // Let sqlContext know which table you gonna work with
    df.registerTempTable("Customers")
    df.show(5)
  }
}

Nếu thành công, bạn sẽ nhận được output như bên dưới.

+----------+--------------------+------------------+--------------------+-----------+----------+-------+
|CustomerID|        CustomerName|       ContactName|             Address|       City|PostalCode|Country|
+----------+--------------------+------------------+--------------------+-----------+----------+-------+
|         1| Alfreds Futterkiste|      Maria Anders|       Obere Str. 57|     Berlin|     12209|Germany|
|         2|Ana Trujillo Empa...|      Ana Trujillo|Avda. de la Const...|México D.F.|      5021| Mexico|
|         3|Antonio Moreno Ta...|    Antonio Moreno|      Mataderos 2312|México D.F.|      5023| Mexico|
|         4|     Around the Horn|      Thomas Hardy|     120 Hanover Sq.|     London|   WA1 1DP|     UK|
|         5|  Berglunds snabbköp|Christina Berglund|      Berguvsvägen 8|      Luleå|  S-958 22| Sweden|
+----------+--------------------+------------------+--------------------+-----------+----------+-------+
only showing top 5 rows

Ở các phần còn lại, ta sẽ tiến hành các truy vấn cơ bản trên bảng dữ liệu Customers.

Câu lệnh SELECT

Bạn có 2 lựa chọn để có thể truy vấn dữ liệu

1. Sử dụng sqlContext với hàm sql(): bạn sẽ cảm thấy thân thuộc ngay vì đây hoàn toàn là các câu lệnh SQL bạn từng làm việc.
2. Sử dụng DataFrames: kiểu dữ liệu DataFrames cung cấp cho các bạn đầy đủ các hàm tương tự như SQL (SQL-like)  để bạn dễ dàng truy vấn.

Tuỳ theo thói quen và sở thích, bạn có thể sử dụng một trong 2 cách tiếp cận này. Điều quan trọng là kết quả xuất ra đúng như mong đợi.

// SELECT by sqlContext
sqlContext.sql("SELECT CustomerName,City FROM Customers").show(5)
// SELECT by DataFrames
df.select("CustomerName", "City").show(5)

// Output
+--------------------+-----------+
|        CustomerName|       City|
+--------------------+-----------+
| Alfreds Futterkiste|     Berlin|
|Ana Trujillo Empa...|México D.F.|
|Antonio Moreno Ta...|México D.F.|
|     Around the Horn|     London|
|  Berglunds snabbköp|      Luleå|
+--------------------+-----------+
only showing top 5 rows

SQL SELECT DISTINCT Statement

// SELECT DISTINCT by sqlContext
sqlContext.sql("SELECT DISTINCT City FROM Customers").show(5)
// SELECT DISTINCT by DataFrames
df.select("City").distinct().show(5)

// Output
+--------+
|    City|
+--------+
|  Aachen|
|    Bern|
|  Nantes|
|Toulouse|
|   Walla|
+--------+
only showing top 5 rows

SQL WHERE Clause

Scala còn cung cấp cho bạn lựa chọn viết code SQL trên nhiều dòng bằng cách sử dụng dấu nháy 3 “””_”””. Thêm vào đó, bạn còn có thể lựa chọn phong cách viết code theo trường phái Scala hay Java như ví dụ minh hoạ bên dưới. Tôi sẽ phân biệt hai trường phái này thông qua hai hàm where (scala) và filter (java).

// WHERE clause
import org.apache.spark.sql.functions._
sqlContext.sql(
  """
    SELECT * FROM Customers
    WHERE Country='Mexico'
  """).show(5)
// Scala style
df.where("Country='Mexico'").show(5)
// Java style
df.filter(col("Country") === "Mexico").show(5)

// Output
+----------+--------------------+--------------------+--------------------+-----------+----------+-------+
|CustomerID|        CustomerName|         ContactName|             Address|       City|PostalCode|Country|
+----------+--------------------+--------------------+--------------------+-----------+----------+-------+
|         2|Ana Trujillo Empa...|        Ana Trujillo|Avda. de la Const...|México D.F.|      5021| Mexico|
|         3|Antonio Moreno Ta...|      Antonio Moreno|      Mataderos 2312|México D.F.|      5023| Mexico|
|        13|Centro comercial ...|     Francisco Chang|Sierras de Granad...|México D.F.|      5022| Mexico|
|        58|Pericles Comidas ...| Guillermo Fernández|Calle Dr. Jorge C...|México D.F.|      5033| Mexico|
|        80| Tortuga Restaurante|Miguel Angel Paolino|    Avda. Azteca 123|México D.F.|      5033| Mexico|
+----------+--------------------+--------------------+--------------------+-----------+----------+-------+

Nguồn tham khảo:

• SQL Tutorial
• CSV Data Source for Spark
• Supported syntax of Spark SQL
• Spark SQL – DataFrames
• Spark SQL, DataFrames and Datasets Guide

Advertisement