第05章 数据湖技术¶

from delta.tables import DeltaTable
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("DeltaLakeExample") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# 写入Delta表
df = spark.read.json("data.json")
df.write.format("delta").save("/delta/sales")

# 读取Delta表
df = spark.read.format("delta").load("/delta/sales")
df.show()

# 更新Delta表
from delta.tables import DeltaTable
deltaTable = DeltaTable.forPath(spark, "/delta/sales")
deltaTable.update(
    condition = "amount > 1000",
    set = { "amount": "amount * 1.1" }
)

# 删除Delta表数据
deltaTable.delete("amount < 0")

# 时间旅行
df_old = spark.read.format("delta").option("versionAsOf", 0).load("/delta/sales")

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("IcebergExample") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.local", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.local.type", "hadoop") \
    .config("spark.sql.catalog.local.warehouse", "s3://my-bucket/warehouse") \
    .getOrCreate()

# 创建Iceberg表
spark.sql("""
    CREATE TABLE local.db.sales (
        id BIGINT,
        product_id BIGINT,
        customer_id BIGINT,
        sale_date DATE,
        quantity INT,
        amount DECIMAL(10,2)
    ) USING iceberg
    PARTITIONED BY (years(sale_date))
    LOCATION 's3://my-bucket/sales'
""")

# 写入数据（DataFrameWriterV2 API）
df.writeTo("local.db.sales").append()

# 读取数据
df = spark.sql("SELECT * FROM local.db.sales")
df.show()

# 更新数据
spark.sql("""
    UPDATE local.db.sales
    SET amount = amount * 1.1
    WHERE amount > 1000
""")

# 时间旅行
df_old = spark.sql("SELECT * FROM local.db.sales VERSION AS OF 0")

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("HudiExample") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .config("spark.sql.hive.convertMetastoreParquet", "false") \
    .getOrCreate()

# 写入Hudi表
hudi_options = {
    'hoodie.table.name': 'sales',
    'hoodie.datasource.write.recordkey.field': 'sale_id',
    'hoodie.datasource.write.precombine.field': 'sale_date',
    'hoodie.datasource.write.table.type': 'COPY_ON_WRITE',
    'hoodie.upsert.shuffle.parallelism': 2,
    'hoodie.insert.shuffle.parallelism': 2
}

df.write.format("org.apache.hudi") \
    .options(**hudi_options) \
    .mode("append") \
    .save("s3://my-bucket/sales")

# 读取Hudi表
df = spark.read.format("org.apache.hudi") \
    .load("s3://my-bucket/sales/*")
df.show()

# 增量写入
df.write.format("org.apache.hudi") \
    .options(**hudi_options) \
    .option("operation", "upsert") \
    .mode("append") \
    .save("s3://my-bucket/sales")