Hướng dẫn sử dụng Apache Spark từ cơ bản đến nâng cao

1. Apache Spark là gì?

Apache Spark được định nghĩa là một công cụ thống nhất dành cho phân tích dữ liệu quy mô lớn (Unified engine for large-scale data analytics). Khi làm việc với các tập dữ liệu có dung lượng hàng trăm GB, việc sử dụng các công cụ xử lý dữ liệu phân tán trở nên bắt buộc.

Apache Spark là một engine đa ngôn ngữ được thiết kế để thực thi các tác vụ về kỹ thuật dữ liệu (data engineering), khoa học dữ liệu (data science) và học máy (machine learning) trên cả máy đơn node lẫn cluster máy tính. Sự linh hoạt này cho phép người dùng chuyển đổi liền mạch giữa các giai đoạn phát triển, gỡ lỗi và triển khai. Đặc biệt, trong các bài toán phân tích dữ liệu không giám sát như phân tích hành vi người dùng để dự đoán xu hướng, Spark là một lựa chọn lý tưởng.

2. Các công cụ thay thế cho Apache Spark

• Apache Hadoop
• Google BigQuery
• Amazon EMR
• IBM Analytics Engine
• Apache Flink
• Lumify
• Presto
• Apache Pig

3. Tại sao nên chọn Apache Spark?

• Khả năng tính toán phân tán quy mô lớn
• Mã nguồn mở, miễn phí
• Hỗ trợ các nhu cầu phân tích dữ liệu nâng cao
• Tích hợp các module cho machine learning, xử lý stream, SQL và xử lý đồ thị
• Hỗ trợ nhiều ngôn ngữ lập trình: Python, R, Java
• Cộng đồng phát triển tích cực và năng động

4. Cài đặt Apache Spark

4.1. So sánh các phương pháp cài đặt

Các phương pháp cài đặt khả thi:

• Windows 10
• Windows Subsystem Linux (WSL)
• Ubuntu / CentOS
• Docker

Phương pháp khuyến nghị: Docker

Phân tích ưu nhược điểm:

• Windows 10: Không phù hợp cho phát triển ứng dụng do thiếu công cụ dòng lệnh, nhiều vấn đề ẩn và tài liệu hỗ trợ hạn chế
• WSL: Cần cài đặt nhiều phần mềm và cấu hình biến môi trường, quy trình phức tạp
• Ubuntu/CentOS: Tương tự WL nhưng cài trực tiếp trên máy
• Docker: Đơn giản, hiệu quả và dễ di chuyển giữa các môi trường

4.2. Hướng dẫn cài đặt qua Docker

Bước 1: Cài đặt Docker (đã có sẵn tài liệu hướng dẫn riêng)

Bước 2: Kéo image Spark:

docker pull jupyter/pyspark-notebook

Bước 3: Tạo container:

docker run \
    -d \
    -p 8022:22 \
    -p 4040:4040 \
    -p 4041:4041 \
    -p 8888:8888 \
    -v /home/userdata:/data \
    -e GRANT_SUDO=yes \
    --name spark_env \
    jupyter/pyspark-notebook

Bước 4: Cấu hình SSH cho container:

docker exec \
    -u 0 \
    -it \
    spark_env \
    /bin/bash

Cài đặt các phần mềm cần thiết:

apt update && apt install openssh-server htop tmux python3-pip

Cấu hình đăng nhập root qua SSH:

echo "PermitRootLogin yes" >> /etc/ssh/sshd_config
service ssh --full-restart
passwd root

Kiểm tra kết nối SSH:

ssh root@127.0.0.1 -p 8022

Bước 5: Cấu hình môi trường Python trong container:

pip3 install pyspark numpy pandas tqdm scikit-learn

Kiểm tra cài đặt:

python3 /usr/local/spark/examples/src/main/python/pi.py

4.3. Kiểm tra cài đặt

Truy cập thư mục gốc của Spark:

cd /usr/local/spark

Kiểm tra tính năng tính toán phân tán với bài toán tính số PI:

python3 examples/src/main/python/pi.py

Kết quả hiển thị thông báo sau đây cho thấy cluster hoạt động bình thường:

Pi is roughly 3.130000

Kiểm tra thuật toán ALS (Alternating Least Square) - một thuật toán phân rã ma trận có thể song song hóa, phù hợp cho các bài toán collaborative filtering quy mô lớn:

python3 examples/src/main/python/als.py

Kết quả đầu ra:

Running ALS with M=100, U=500, F=10, iters=5, partitions=2

Iteration 0:                                                                            
RMSE: 0.2229

Iteration 1:
RMSE: 0.0731

Iteration 2:
RMSE: 0.0317

Iteration 3:
RMSE: 0.0315

Iteration 4:
RMSE: 0.0315

Kiểm tra tính năng giám sát trạng thái cluster:

python3 examples/src/main/python/status_api_demo.py

Kiểm tra tính toán đồ thị - thuật toán tìm bao đóng bắc cầu (transitive closure):

python3 examples/src/main/python/transitive_closure.py

Kết quả:

TC has 6816 edges

5. Thực hành với Spark

Kiến trúc chương trình Spark bao gồm các module chính sau:

Tên module	Mô tả
RDD	Resilient Distributed Datasets, accumulators và broadcast variables
Spark SQL, Datasets, DataFrames	Xử lý dữ liệu có cấu trúc với truy vấn quan hệ
Structured Streaming	Xử lý stream dữ liệu có cấu trúc
Spark Streaming	Xử lý stream sử dụng DStreams
MLlib	Thư viện học máy
GraphX	Xử lý đồ thị
PySpark	Giao diện Python cho Spark

5.1. Thực hành với DataFrames

Bài tập 1: Xử lý dữ liệu với PySpark

from pyspark.sql import SparkSession


# Khởi tạo kết nối Spark
spark = SparkSession \
    .builder \
    .appName('data_processor') \
    .getOrCreate()

# Tạo DataFrame mẫu
data = [
    ('stu001','grade1',100,87,67,83,98),
    ('stu002','grade2',87,81,90,83,83),
    ('stu003','grade3',86,91,83,89,63),
    ('stu004','grade2',65,87,94,73,88),
    ('stu005','grade1',76,62,89,81,98),
    ('stu006','grade3',84,82,85,73,99),
    ('stu007','grade3',56,76,63,72,87),
    ('stu008','grade1',55,62,46,78,71),
    ('stu009','grade2',63,72,87,98,64)]

dataframe = spark.createDataFrame(
    data,
    ['student_id','grade','literature','math','english','physics','chemistry'])

dataframe.show()

# Đếm số dòng và cột
print("Số dòng: %d" % dataframe.count())
print("Số cột: %d" % len(dataframe.columns))

# Lọc học sinh lớp grade1
dataframe.filter("grade = 'grade1'").show(n=3)

# Lọc học sinh có điểm literature > 90 hoặc math > 90
dataframe.filter("literature > 90 or math > 90").show(n=3)

# Đóng kết nối
spark.stop()

Bài tập 2: Thống kê dữ liệu

from pyspark import SparkFiles
from pyspark.sql import SparkSession


# Kết nối Spark
spark = SparkSession \
    .builder \
    .appName('statistics') \
    .getOrCreate()

# Đọc file dữ liệu
spark.sparkContext.addFile("https://cdn.coggle.club/Pokemon.csv")
file_path = "file://" + SparkFiles.get("Pokemon.csv")

# Load dữ liệu với header
df = spark.read.csv(path=file_path, header=True, inferSchema=True)
df.show(n=3)

# Đổi tên cột có dấu chấm
df = df.withColumnRenamed('Sp. Atk', 'Sp_Atk')
df = df.withColumnRenamed('Sp. Def', 'Sp_Def')

# Phân tích kiểu dữ liệu mỗi cột và số giá trị distinct
for col_name, col_type in df.dtypes:
    print((col_name, col_type))
    df.groupBy(col_name).count().show(n=3)

# Kiểm tra giá trị null
for col_name in df.columns:
    null_count = df.filter(df[col_name].isNull()).count()
    print("Cột: %s \t ,Số giá trị null: %d" % (col_name, null_count))

spark.stop()

Bài tập 3: Group và Aggregate

from pyspark.sql import SparkSession
from pyspark.sql import functions as func
from pyspark import SparkFiles


# Kết nối cluster
spark = SparkSession \
    .builder \
    .appName('aggregation') \
    .getOrCreate()

# Load dữ liệu
spark.sparkContext.addFile("https://cdn.coggle.club/Pokemon.csv")
path = "file://" + SparkFiles.get("Pokemon.csv")
df = spark.read.csv(path=path, header=True, inferSchema=True)
df = df.withColumnRenamed('Sp. Atk', 'Sp_Atk')
df = df.withColumnRenamed('Sp. Def', 'Sp_Def')
df.show(n=3)

# Sử dụng groupBy
df.groupBy("Type 1").count().show(n=3)

# Sử dụng agg với các hàm thống kê
df.agg(func.max("HP")).show()
df.agg(func.min("HP")).show()
df.agg(func.avg("HP")).show()

# Sử dụng transform
def display_info(input_df):
    print(type(input_df))
    input_df.show(n=3)
    return input_df

df.transform(display_info)

# Kết hợp groupBy, agg để tính trung bình HP theo Type 1
df.groupBy("Type 1").agg(func.avg("HP")).show(n=3)

spark.stop()

Bài tập 4: Cú pháp SparkSQL cơ bản

from pyspark import SparkFiles
from pyspark.sql import SparkSession


# Kết nối cluster
spark = SparkSession.builder.appName('spark_sql').getOrCreate()

# Tạo DataFrame mẫu
student_data = [
    ('stu001','grade1',100,87,67,83,98),
    ('stu002','grade2',87,81,90,83,83),
    ('stu003','grade3',86,91,83,89,63),
    ('stu004','grade2',65,87,94,73,88),
    ('stu005','grade1',76,62,89,81,98),
    ('stu006','grade3',84,82,85,73,99),
    ('stu007','grade3',56,76,63,72,87),
    ('stu008','grade1',55,62,46,78,71),
    ('stu009','grade2',63,72,87,98,64)]

df = spark.createDataFrame(
    student_data,
    ['student_id','grade','literature','math','english','physics','chemistry'])

# Đăng ký bảng ảo
df.createTempView("students")

# Đếm số dòng
result = spark.sql("SELECT COUNT(*) FROM students")
result.show()

# Đếm số cột
col_info = spark.sql("SHOW COLUMNS IN students")
col_info.show()
print(col_info.count())

# Lọc grade = grade1
result = spark.sql("SELECT * FROM students WHERE grade='grade1'")
result.show(n=3)

# Lọc literature > 90 hoặc math > 90
result = spark.sql("SELECT * FROM students WHERE literature>90 OR math>90")
result.show(n=3)

# Đọc file Pokemon.csv
spark.sparkContext.addFile("https://cdn.coggle.club/Pokemon.csv")
path = "file://" + SparkFiles.get("Pokemon.csv")

df2 = spark.read.csv(path=path, header=True, inferSchema=True)
df2 = df2.withColumnRenamed('Sp. Atk', 'SpAtk')
df2 = df2.withColumnRenamed('Sp. Def', 'SpDef')
df2 = df2.withColumnRenamed('Type 1', 'Type1')
df2 = df2.withColumnRenamed('Type 2', 'Type2')
df2.createOrReplaceTempView("pokemon")

# Phân tích kiểu dữ liệu
schema_info = spark.sql("DESCRIBE pokemon")
schema_info.show()

# Đếm giá trị distinct
distinct_count = spark.sql("SELECT COUNT(DISTINCT Attack) FROM pokemon")
distinct_count.show(n=3)

# Đếm distinct cho mỗi cột
for col_name in df2.schema.names:
    print("Cột %s có %d giá trị distinct" % (col_name, df2.select(col_name).distinct().count()))

# Kiểm tra null
null_total = spark.sql("SELECT COUNT(*) FROM pokemon")
null_total.show()

type2_count = spark.sql("SELECT COUNT(Type2) FROM pokemon")
type2_count.show()

# Group by và tính trung bình HP theo Type 1
avg_hp = spark.sql("SELECT Type1, AVG(HP) FROM pokemon GROUP BY Type1")
avg_hp.show()

spark.stop()

Bài tập 5: SparkML - Mã hóa dữ liệu

from pyspark.sql import SparkSession
from pyspark import SparkFiles
from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer
from pyspark.ml.feature import OneHotEncoder
from pyspark.ml.feature import MinMaxScaler
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.feature import PCA


# Kết nối cluster
spark = SparkSession.builder.appName('feature_engineering').getOrCreate()

# Load dữ liệu Pokemon
spark.sparkContext.addFile("https://cdn.coggle.club/Pokemon.csv")
path = "file://" + SparkFiles.get("Pokemon.csv")

df = spark.read.csv(path=path, header=True, inferSchema=True)
df = df.withColumnRenamed('Sp. Atk', 'SpAtk')
df = df.withColumnRenamed('Sp. Def', 'SpDef')
df = df.withColumnRenamed('Type 1', 'Type1')
df = df.withColumnRenamed('Type 2', 'Type2')
df.show(n=3)

# Cột categorical: Type1, Type2, Generation
# Cột numeric: Total, HP, Attack, Defense, SpAtk, SpDef, Speed

# Bước 1: String Indexer cho các cột categorical
indexer = StringIndexer(
    inputCols=["Type1", "Type2"],
    outputCols=["Type1_indexed", "Type2_indexed"],
    handleInvalid='skip')
df = indexer.fit(df).transform(df)
df.show(n=3)

# Bước 2: One-Hot Encoding
encoder = OneHotEncoder(
    inputCols=['Type1_indexed', 'Type2_indexed', 'Generation'],
    outputCols=["Type1_encoded", "Type2_encoded", "Generation_encoded"])
df = encoder.fit(df).transform(df)
df.show(n=3)

# Bước 3: MinMaxScaler cho các cột numeric
numeric_cols = ["Total", "HP", "Attack", "Defense", "SpAtk", "SpDef", "Speed"]
vector_assemblers, scalers = [], []

for col in numeric_cols:
    assembler = VectorAssembler(inputCols=[col], outputCol=col + "_vec")
    vector_assemblers.append(assembler)
    scaler = MinMaxScaler(inputCol=col + "_vec", outputCol=col + "_normalized")
    scalers.append(scaler)

pipeline = Pipeline(stages=vector_assemblers + scalers)
df = pipeline.fit(df).transform(df)
df.show(n=3)

# Bước 4: PCA dimensionality reduction
feature_cols = ["Type1_encoded", "Type2_encoded", "Generation_encoded", 
                "Total_normalized", "HP_normalized", "Attack_normalized", 
                "Defense_normalized", "SpAtk_normalized", "SpDef_normalized", "Speed_normalized"]

assembler = VectorAssembler(inputCols=feature_cols, outputCol="features")
df = assembler.transform(df)
df.select("features").show(n=3)

# Áp dụng PCA với k=5
pca = PCA(k=5, inputCol="features", outputCol="pca_features")
df = pca.fit(df).transform(df)
df.show(n=3)

results = df.select("pca_features").collect()
print(results[0].asDict())

spark.stop()

Bài tập 6: SparkML - Mô hình phân loại

from pyspark import SparkFiles
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.ml.feature import StringIndexer
from pyspark.ml.feature import MinMaxScaler
from pyspark.ml.feature import VectorAssembler
from pyspark.ml import Pipeline
from pyspark.ml.classification import DecisionTreeClassifier
from pyspark.ml.classification import RandomForestClassifier
from pyspark.ml.classification import NaiveBayes
from pyspark.ml.evaluation import MulticlassClassificationEvaluator


# Khởi tạo Spark
spark = SparkSession.builder.appName('classification').getOrCreate()

# Load dữ liệu
spark.sparkContext.addFile("https://cdn.coggle.club/Pokemon.csv")
path = "file://" + SparkFiles.get("Pokemon.csv")

df = spark.read.csv(path=path, header=True, inferSchema=True)
df = df.withColumnRenamed('Sp. Atk', 'SpAtk')
df = df.withColumnRenamed('Sp. Def', 'SpDef')
df = df.withColumnRenamed('Type 1', 'Type1')
df = df.withColumnRenamed('Type 2', 'Type2')
df = df.withColumn("Legendary", col("Legendary").cast('string'))

# Label encoding cho Type1 (label)
label_indexer = StringIndexer(inputCol="Type1", outputCol="label")
df = label_indexer.fit(df).transform(df)

# Encoding các feature categorical
cat_input = ["Type2", "Generation", "Legendary"]
cat_output = ["Type2_idx", "Generation_idx", "Legendary_idx"]
cat_indexer = StringIndexer(inputCols=cat_input, outputCols=cat_output, handleInvalid="skip")
df = cat_indexer.fit(df).transform(df)

# Encoding các feature numeric
numeric_features = ["Total", "HP", "Attack", "Defense", "SpAtk", "SpDef", "Speed"]
assemblers, scalers_list = [], []

for col_name in numeric_features:
    vec_assembler = VectorAssembler(inputCols=[col_name], outputCol=col_name + "_vec")
    assemblers.append(vec_assembler)
    scaler = MinMaxScaler(inputCol=col_name + "_vec", outputCol=col_name + "_scaled")
    scalers_list.append(scaler)

norm_pipeline = Pipeline(stages=assemblers + scalers_list)
df = norm_pipeline.fit(df).transform(df)

# Tạo feature vector
final_features = ["Type2_idx", "Generation_idx", "Legendary_idx",
                  "Total_scaled", "HP_scaled", "Attack_scaled", 
                  "Defense_scaled", "SpAtk_scaled", "SpDef_scaled", "Speed_scaled"]

feature_assembler = VectorAssembler(inputCols=final_features, outputCol="feature_vector")
df = feature_assembler.transform(df)

# Chia train/test
train_data, test_data = df.randomSplit(weights=[0.8, 0.2], seed=123)

# Đánh giá
evaluator = MulticlassClassificationEvaluator(
    labelCol="label",
    predictionCol="prediction",
    metricName="accuracy")

# Định nghĩa các mô hình
classifiers = {
    "Decision Tree": DecisionTreeClassifier(labelCol="label", featuresCol="feature_vector", predictionCol="prediction"),
    "Random Forest": RandomForestClassifier(labelCol="label", featuresCol="feature_vector", predictionCol="prediction"),
    "Naive Bayes": NaiveBayes(labelCol="label", featuresCol="feature_vector", predictionCol="prediction")
}

# Huấn luyện và đánh giá
for name, model in classifiers.items():
    predictions = model.fit(train_data).transform(test_data)
    accuracy = evaluator.evaluate(predictions)
    print("Độ chính xác %s: %.4f" % (name, accuracy))

spark.stop()

Bài tập 7: SparkML - Mô hình phân cụm

from pyspark import SparkFiles
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.ml.feature import StringIndexer
from pyspark.ml.feature import MinMaxScaler
from pyspark.ml.feature import VectorAssembler
from pyspark.ml import Pipeline
from pyspark.ml.clustering import KMeans
from pyspark.sql.types import DoubleType
from pyspark.ml.evaluation import MulticlassClassificationEvaluator


# Khởi tạo Spark
spark = SparkSession.builder.appName('clustering').getOrCreate()

# Load dữ liệu
spark.sparkContext.addFile("https://cdn.coggle.club/Pokemon.csv")
path = "file://" + SparkFiles.get("Pokemon.csv")

df = spark.read.csv(path=path, header=True, inferSchema=True)
df = df.withColumnRenamed('Sp. Atk', 'SpAtk')
df = df.withColumnRenamed('Sp. Def', 'SpDef')
df = df.withColumnRenamed('Type 1', 'Type1')
df = df.withColumnRenamed('Type 2', 'Type2')
df = df.withColumn("Legendary", col("Legendary").cast('string'))

# Label encoding cho Type1
type1_indexer = StringIndexer(inputCol="Type1", outputCol="type1_label")
df = type1_indexer.fit(df).transform(df)

# Encoding categorical features
categorical_cols = ["Type2", "Generation", "Legendary"]
categorical_outputs = ["Type2_labeled", "Generation_labeled", "Legendary_labeled"]

indexer_stage = StringIndexer(inputCols=categorical_cols, outputCols=categorical_outputs, handleInvalid="skip")
df = indexer_stage.fit(df).transform(df)

# Encoding numerical features
numerical_cols = ["Total", "HP", "Attack", "Defense", "SpAtk", "SpDef", "Speed"]
vec_assemblers, minmax_scalers = [], []

for col_item in numerical_cols:
    assembler = VectorAssembler(inputCols=[col_item], outputCol=col_item + "_vector")
    vec_assemblers.append(assembler)
    scaler = MinMaxScaler(inputCol=col_item + "_vector", outputCol=col_item + "_scaled")
    minmax_scalers.append(scaler)

preprocessing_pipeline = Pipeline(stages=vec_assemblers + minmax_scalers)
df = preprocessing_pipeline.fit(df).transform(df)

# Tạo feature vector cuối cùng
all_features = ["Type2_labeled", "Generation_labeled", "Legendary_labeled",
                "Total_scaled", "HP_scaled", "Attack_scaled", 
                "Defense_scaled", "SpAtk_scaled", "SpDef_scaled", "Speed_scaled"]

final_assembler = VectorAssembler(inputCols=all_features, outputCol="final_features")
df = final_assembler.transform(df)

# Chia dữ liệu
train_set, test_set = df.randomSplit(weights=[0.8, 0.2], seed=123)

# Evaluator
cluster_evaluator = MulticlassClassificationEvaluator(
    labelCol="type1_label",
    predictionCol="prediction",
    metricName="accuracy")

# Tìm số cụm tối ưu bằng phương pháp Elbow
unique_type1_count = df.select("Type1").distinct().count()

for k in range(2, unique_type1_count + 1):
    kmeans = KMeans(featuresCol="final_features", predictionCol="prediction", k=k, seed=123)
    trained_model = kmeans.fit(train_set)
    predictions = trained_model.transform(test_set)
    predictions = predictions.withColumn("prediction", predictions.prediction.cast(DoubleType()))
    
    cost = trained_model.summary.trainingCost
    accuracy = cluster_evaluator.evaluate(predictions)
    print("Với k=%d: Độ chính xác = %.4f, Chi phí = %.4f" % (k, accuracy, cost))

spark.stop()

Tài liệu tham khảo

[1] Apache Spark. https://spark.apache.org/