在 PySpark 和 PyFlink 中写入 Hive 表，本质上是利用这些引擎的 Writer（写入器） 按照 Hive 的元数据规范（分区、存储格式、序列化方式）将数据生成文件，并更新 Hive Metastore。
以下是具体的实现代码与核心配置：
1. PySpark 写入 Hive 表
PySpark 写入 Hive 非常直观，主要使用 saveAsTable 方法。
写入模式 (SaveMode)
-
overwrite: 覆盖表或分区。 -
append: 追加数据。 -
errorifexists: 表存在则报错（默认）。
代码示例
from pyspark.sql import SparkSessionspark = SparkSession.builder \.appName("PySpark_Write_Hive") \.enableHiveSupport() \.getOrCreate()# 创建一个 DataFrame
data = [("Alice", 28), ("Bob", 35)]
df = spark.createDataFrame(data, ["name", "age"])# 1. 简单写入（如果表不存在会自动创建，默认格式通常为 Parquet）
df.write.mode("overwrite").saveAsTable("my_db.user_info")# 2. 写入分区表（强烈建议，提高查询性能）
# 注意：写入前需确保开启动态分区配置
spark.sql("SET hive.exec.dynamic.partition.mode=nonstrict")df.write \.partitionBy("age") \.format("orc") \.mode("append") \.saveAsTable("my_db.user_partitioned")
2. PyFlink 写入 Hive 表
PyFlink 写入 Hive 需要依赖 Hive Catalog 和 Batch/Streaming 模式转换。
核心机制
PyFlink 写入 Hive 时，如果是流式写入（如从 Kafka 到 Hive），它会自动处理小文件合并（Compaction）和分区提交。
代码示例
from pyflink.table import TableEnvironment, EnvironmentSettings, HiveCatalogsettings = EnvironmentSettings.new_instance().in_batch_mode().build()
t_env = TableEnvironment.create(settings)# 注册 Hive Catalog
catalog = HiveCatalog("my_hive", "default", "/path/to/hive/conf")
t_env.register_catalog("my_hive", catalog)
t_env.use_catalog("my_hive")# 创建一个临时表或获取数据源
t_env.execute_sql("CREATE TEMPORARY VIEW source_view AS ...")# 1. 使用 SQL 写入
t_env.execute_sql("INSERT OVERWRITE my_db.target_table SELECT name, age FROM source_view")# 2. 使用 Table API 写入
table = t_env.from_path("source_view")
table.execute_insert("my_db.target_table", overwrite=True)
3. 关键配置与性能优化
写入 Hive 时，最常见的问题是小文件过多或权限报错。请注意以下配置：
A. 动态分区配置
如果你在往分区表写数据，必须开启以下参数，否则会报 Dynamic partition strict mode 错误：
-
Spark:
spark.sql("SET hive.exec.dynamic.partition.mode=nonstrict") -
Flink:
SET 'table.exec.hive.dynamic-partition-writing.enabled' = 'true'
B. 文件合并（避免小文件）
-
Spark: 写入前使用
.repartition(num)控制输出文件的数量，使其与分区数匹配。 -
Flink: 开启自动合并功能：
-- 仅限 Flink 1.13+ SET 'sink.partition-commit.policy.kind' = 'metastore,success-file'; SET 'auto-compaction' = 'true';
C. 存储格式一致性
确保代码中的 format 与 Hive 表定义的 STORED AS 一致。例如，Hive 表定义为 ORC，Spark 写入时也应指定 .format("orc")。
4. 总结对比
| 特性 | PySpark | PyFlink |
| 易用性 | 极高，saveAsTable 一行代码搞定 |
中等，需配置 Catalog 和 Environment |
| 实时性 | 侧重离线批量写入 | 侧重实时流式入湖 (Streaming Sink) |
| 分区管理 | 灵活，支持静态和动态分区 | 强大，支持基于时间的自动分区提交 |
| 适用场景 | 每日定时 ETL 任务 | 实时数据同步（如 MySQL 到 Hive） |