Delta Lake Home Lab Setup — สร้าง Data Lakehouse

Delta Lake Home Lab

Delta Lake Home Lab Setup Data Lakehouse Spark MinIO ACID Transactions Time Travel Schema Enforcement Pipeline Bronze Silver Gold

Docker Compose Setup

=== Delta Lake Home Lab Docker Compose ===

docker-compose.yml

version: '3.8'

services:

spark-master:

image: bitnami/spark:3.5

environment:

• SPARK_MODE=master
• SPARK_MASTER_HOST=spark-master

ports:

• "8080:8080" # Spark UI
• "7077:7077" # Spark Master

volumes:

• ./spark-defaults.conf:/opt/bitnami/spark/conf/spark-defaults.conf

spark-worker-1:

image: bitnami/spark:3.5

environment:

• SPARK_MODE=worker
• SPARK_MASTER_URL=spark://spark-master:7077
• SPARK_WORKER_MEMORY=4G
• SPARK_WORKER_CORES=2

minio:

image: minio/minio

command: server /data --console-address ":9001"

ports:

• "9000:9000" # S3 API
• "9001:9001" # Console

environment:

MINIO_ROOT_USER: minioadmin

MINIO_ROOT_PASSWORD: minioadmin

volumes:

• minio-data:/data

jupyter:

image: jupyter/pyspark-notebook

ports:

• "8888:8888"

environment:

• SPARK_MASTER=spark://spark-master:7077

volumes:

• ./notebooks:/home/jovyan/work

spark-defaults.conf

spark.jars.packages io.delta:delta-spark_2.12:3.0.0, org.apache.hadoop:hadoop-aws:3.3.4

spark.sql.extensions io.delta.sql.DeltaSparkSessionExtension

spark.sql.catalog.spark_catalog org.apache.spark.sql.delta.catalog.DeltaCatalog

spark.hadoop.fs.s3a.endpoint http://minio:9000

spark.hadoop.fs.s3a.access.key minioadmin

spark.hadoop.fs.s3a.secret.key minioadmin

spark.hadoop.fs.s3a.path.style.access true

spark.hadoop.fs.s3a.impl org.apache.hadoop.fs.s3a.S3AFileSystem

from dataclasses import dataclass

@dataclass

class LabComponent:

component: str

image: str

เนื้อหาเกี่ยวข้อง — บทความที่เกี่ยวข้อง: Midjourney Prompt Infrastructure as Code

port: str

resource: str

config_tip: str

components = [

LabComponent("Spark Master",

"bitnami/spark:3.5",

"8080 (UI) 7077 (Master)",

แนะนำเพิ่มเติม — แหล่งความรู้ Forex iCafeForex

"2 Core 2GB RAM",

"ตั้ง SPARK_DAEMON_MEMORY=1g สำหรับ Home Lab"),

LabComponent("Spark Worker",

"bitnami/spark:3.5",

"8081 (Worker UI)",

"2 Core 4GB RAM ต่อ Worker",

"เริ่ม 1-2 Workers ตามเครื่อง"),

LabComponent("MinIO",

"minio/minio",

"9000 (S3) 9001 (Console)",

"1 Core 1GB RAM SSD",

"ใช้ SSD ไม่ใช่ HDD สำหรับ Performance"),

LabComponent("Jupyter",

"jupyter/pyspark-notebook",

"8888",

"1 Core 2GB RAM",

"Install delta-spark ใน Notebook"),

เนื้อหาเกี่ยวข้อง — ทำความเข้าใจ What is Resistance and Support —

]

Data Pipeline (Medallion)

=== Medallion Architecture Pipeline ===

from pyspark.sql import SparkSession

from delta.tables import DeltaTable

spark = SparkSession.builder \
.appName("DeltaLakePipeline") \
.config("spark.jars.packages", "io.delta:delta-spark_2.12:3.0.0") \

.getOrCreate()

# Bronze: Raw Ingest

แนะนำเพิ่มเติม — SiamCafeBook

raw_df = spark.read.json("s3a://datalake/raw/events/")

raw_df.write.format("delta").mode("append") \

.save("s3a://datalake/bronze/events")

# Silver: Clean + Transform

bronze_df = spark.read.format("delta").load("s3a://datalake/bronze/events")

silver_df = bronze_df \
.dropna(subset=["user_id", "event_type"]) \
.dropDuplicates(["event_id"]) \

.withColumn("event_date", F.to_date("timestamp"))

silver_df.write.format("delta").mode("overwrite") \

เนื้อหาเกี่ยวข้อง — บทความที่เกี่ยวข้อง: financial freedom pdf

.save("s3a://datalake/silver/events")

# Gold: Aggregate

silver_df = spark.read.format("delta").load("s3a://datalake/silver/events")

gold_df = silver_df.groupBy("event_date", "event_type") \

.agg(F.count("*").alias("event_count"),

F.countDistinct("user_id").alias("unique_users"))

gold_df.write.format("delta").mode("overwrite") \

.save("s3a://datalake/gold/daily_metrics")

# Time Travel

df_v0 = spark.read.format("delta").option("versionAsOf", 0) \

.load("s3a://datalake/silver/events")

# MERGE (Upsert)

delta_table = DeltaTable.forPath(spark, "s3a://datalake/silver/users")

delta_table.alias("target").merge(

new_data.alias("source"),

"target.user_id = source.user_id"

).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()

@dataclass

class PipelineLayer:

layer: str

data: str

operations: str

schedule: str

retention: str

layers = [

เนื้อหาเกี่ยวข้อง — การนํา smart contract

PipelineLayer("Bronze (Raw)",

"Raw Data as-is จาก Source",

"Ingest Append Only ไม่แปลง",

"Real-time (Streaming) หรือ ทุก 15 นาที",

"90 วัน (VACUUM)"),

PipelineLayer("Silver (Cleaned)",

"Cleaned Validated Deduplicated",

"dropna dropDuplicates Cast Schema Enforce",

"ทุก 1 ชั่วโมง หรือ Triggered",

"1 ปี"),

PipelineLayer("Gold (Aggregated)",

"Business Metrics KPI Summary",

"groupBy agg join Materialized View",

"ทุกวัน หรือ ทุก 1 ชั่วโมง",

"ถาวร"),

]

เคล็ดลับ

• SSD: ใช้ SSD สำหรับ MinIO + Spark Shuffle ดีกว่า HDD มาก
• ZORDER: ใช้ ZORDER กับ Column ที่ Filter บ่อย
• Partition: Partition ตาม Date สำหรับ Time Series Data
• VACUUM: ตั้ง VACUUM ทุกสัปดาห์ ลด Storage
• Docker: เริ่ม Docker Compose ง่ายที่สุดสำหรับ Home Lab

Delta Lake คืออะไร

Open Source Storage Layer Lakehouse ACID Parquet Transaction Log Time Travel Schema Enforcement MERGE OPTIMIZE VACUUM Spark Databricks