Data Lake Architecture: Iceberg trong Crypto Quantitative Trading

Giới thiệu về Apache Iceberg

Apache Iceberg là định dạng bảng mở được thiết kế cho các data lake doanh nghiệp lớn. Iceberg hỗ trợ ACID transactions, time travel queries, và schema evolution. Trong bài viết này, tôi sẽ chia sẻ cách triển khai Iceberg cho hệ thống quantitative trading với dữ liệu crypto.

from pyspark import SparkContext
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

spark = SparkSession.builder \
    .appName("CryptoIcebergPipeline") \
    .config("spark.sql.catalog.my_catalog", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.my_catalog.warehouse", "s3://crypto-data-lake/warehouse/") \
    .config("spark.sql.catalog.my_catalog.s3.endpoint", "https://minio.local:9000") \
    .getOrCreate()

schema = StructType([
    StructField("timestamp", TimestampType(), False),
    StructField("symbol", StringType(), False),
    StructField("open", DoubleType(), False),
    StructField("high", DoubleType(), False),
    StructField("low", DoubleType(), False),
    StructField("close", DoubleType(), False),
    StructField("volume", DoubleType(), False)
])

df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "crypto-ohlcv") \
    .load()

df.writeStream \
    .format("iceberg") \
    .outputMode("append") \
    .option("checkpointLocation", "/tmp/checkpoints/crypto_ohlcv") \
    .toTable("my_catalog.crypto_trading.ohlcv_1m")

Kiến trúc Data Lake cho Crypto Trading

Với dữ liệu crypto có tính biến động cao, kiến trúc Iceberg cung cấp: - **Partitioning thông minh**: Phân chia theo ngày và symbol - **Hidden partitioning**: Tự động filter partition - **Snapshot isolation**: Đọc nhất quán even khi có write đồng thời

from pyspark.sql import functions as F

spark.sql("""
    CREATE TABLE my_catalog.crypto_trading.ohlcv_1m (
        timestamp TIMESTAMP NOT NULL,
        symbol STRING NOT NULL,
        open DOUBLE NOT NULL,
        high DOUBLE NOT NULL,
        low DOUBLE NOT NULL,
        close DOUBLE NOT NULL,
        volume DOUBLE NOT NULL
    )
    USING iceberg
    PARTITIONED BY (days(timestamp), bucket(16, symbol))
    TBLPROPERTIES (
        'write.distribution-mode'='hash',
        'write.metadata.delete-after-commit.enabled'='true',
        'write.metadata.previous-versions-max'='100'
    )
""")

result = spark.sql("""
    SELECT 
        symbol,
        date_format(timestamp, 'yyyy-MM-dd') as trade_date,
        AVG(close) as avg_price,
        SUM(volume) as total_volume
    FROM my_catalog.crypto_trading.ohlcv_1m
    WHERE timestamp >= '2024-01-01' 
        AND symbol IN ('BTCUSDT', 'ETHUSDT')
    GROUP BY symbol, date_format(timestamp, 'yyyy-MM-dd')
    ORDER BY trade_date DESC
""")

result.coalesce(1).write \
    .mode("overwrite") \
    .partitionBy("trade_date") \
    .parquet("s3://crypto-reports/daily_summary/")

Time Travel Queries cho Backtesting

Iceberg cho phép truy vấn data tại bất kỳ thời điểm nào trong quá khứ - critical cho backtesting quantitative strategies.

from datetime import datetime, timedelta

snapshot_id = spark.sql("""
    SELECT snapshot_id 
    FROM my_catalog.crypto_trading.ohlcv_1m.history
    WHERE made_current_at <= '2024-06-15 10:00:00'
    ORDER BY made_current_at DESC
    LIMIT 1
""").collect()[0][0]

historical_df = spark.read \
    .format("iceberg") \
    .option("snapshot-id", snapshot_id) \
    .load("my_catalog.crypto_trading.ohlcv_1m")

strategy_returns = historical_df \
    .withColumn("ma_7d", F.avg("close").over(
        Window.partitionBy("symbol")
              .orderBy(F.col("timestamp").cast("long"))
              .rowsBetween(-7, 0)
    )) \
    .withColumn("ma_25d", F.avg("close").over(
        Window.partitionBy("symbol")
              .orderBy(F.col("timestamp").cast("long"))
              .rowsBetween(-25, 0)
    )) \
    .withColumn("signal", 
        F.when(F.col("ma_7d") > F.col("ma_25d"), 1)
         .when(F.col("ma_7d") < F.col("ma_25d"), -1)
         .otherwise(0)
    )

strategy_returns.createOrReplaceTempView("strategy_signals")

Tối ưu Performance với Iceberg

spark.sql("""
    ALTER TABLE my_catalog.crypto_trading.ohlcv_1m 
    SET TBLPROPERTIES (
        'read.split.target-size'='134217728',
        'read.parquet.vectorization.enabled'='true',
        'read.parquet.vectorization.batch-size'='4096'
    )
""")

spark.sql("OPTIMIZE my_catalog.crypto_trading.ohlcv_1m")
spark.sql("ALTER TABLE my_catalog.crypto_trading.ohlcv_1m COMPACT BINPACK")

Lỗi thường gặp và cách khắc phục

**1. Lỗi "Unable to create table" - Catalog configuration**

org.apache.iceberg.exceptions.RuntimeException: Unable to create table

spark = SparkSession.builder \
    .config("spark.sql.catalog.my_catalog", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.my_catalog.type", "hadoop") \
    .config("spark.sql.catalog.my_catalog.warehouse", "file:///tmp/iceberg_warehouse") \
    .getOrCreate()

**2. Lỗi Concurrent Write Conflict**

org.apache.iceberg.exceptions.CommitFailedException: Cannot commit

df.writeTo("my_catalog.crypto_trading.ohlcv_1m") \
    .option("isolation-level", "serializable") \
    .append()

spark.conf.set("spark.sql.iceberg.write.distribution-mode", "hash")
spark.conf.set("spark.sql.adaptive.enabled", "true")

**3. Lỗi Time Travel với expired snapshots**

org.apache.iceberg.exceptions.ValidationException: Cannot time travel

spark.sql("""
    ALTER TABLE my_catalog.crypto_trading.ohlcv_1m 
    SET TBLPROPERTIES (
        'history.expire.max-snapshot-age-ms'='604800000'
    )
""")

spark.sql("DELETE FROM my_catalog.crypto_trading.ohlcv_1m.snapshots")

Kết luận

Apache Iceberg mang lại reliability và flexibility cần thiết cho data infrastructure trong crypto quantitative trading. Với features như ACID transactions và time travel, việc xây dựng robust backtesting pipeline trở nên đơn giản hơn rất nhiều.