如何在PySpark中创建Dataset？PySpark是否支持该特性？

你说得没错，PySpark确实没有原生支持Dataset API——Dataset是Spark专为JVM语言（Java/Scala）设计的强类型API，依赖静态类型检查机制，而Python作为动态类型语言，无法适配这套强类型约束体系。

不过你可以通过PySpark的DataFrame来实现类似Dataset的功能，因为PySpark的DataFrame在底层执行逻辑上，和Scala/Java中的Dataset[Row]完全等价，都是基于优化后的执行计划运行。以下是几种模拟强类型体验的实用方法：

1. 显式定义结构化Schema #

通过StructType定义严格的Schema，确保数据类型的一致性，这和Dataset的强类型约束效果类似：

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

spark = SparkSession.builder.appName("TypedDataFrame").getOrCreate()

# 定义强类型Schema
user_schema = StructType([
    StructField("username", StringType(), nullable=False),
    StructField("user_id", IntegerType(), nullable=False),
    StructField("email", StringType(), nullable=True)
])

# 基于Schema创建DataFrame
raw_data = [("john_doe", 1001, "john@example.com"), ("jane_smith", 1002, None)]
user_df = spark.createDataFrame(raw_data, schema=user_schema)

user_df.printSchema()
user_df.show()

2. 用Python数据类模拟强类型实体 #

借助Python的dataclasses模块定义结构化数据类，再转换为Spark的Row对象创建DataFrame，类似Scala中用case class定义Dataset的方式：

from dataclasses import dataclass
from pyspark.sql import Row

@dataclass
class Product:
    product_id: int
    product_name: str
    price: float

# 将数据类实例转为Row对象
product_rows = [
    Row(**Product(1, "Laptop", 999.99).__dict__),
    Row(**Product(2, "Mouse", 25.50).__dict__)
]

product_df = spark.createDataFrame(product_rows)
product_df.show()

3. 使用Pandas UDF实现类型安全的函数运算 #

如果需要类似Dataset中强类型UDF的功能，可以使用PySpark的Pandas UDF，它支持类型标注，能实现类型安全的自定义函数：

import pandas as pd
from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import FloatType

# 定义带类型标注的Pandas UDF
@pandas_udf(FloatType())
def calculate_discounted_price(price: pd.Series) -> pd.Series:
    return price * 0.8

# 应用UDF到DataFrame
product_df.withColumn("discounted_price", calculate_discounted_price(product_df["price"])).show()

总结来说，虽然PySpark没有原生的Dataset API，但通过DataFrame配合上述方法，完全可以达到和Dataset一致的使用体验与性能表现。

内容的提问来源于stack exchange，提问作者Jai Barathi