是否存在将PySpark DataFrame写入Azure Cache for Redis的方法？

前置准备 #

• 先获取Azure Cache for Redis的连接参数：在Azure门户进入对应的Redis实例，从「访问密钥」页面获取主机名、主访问密钥、端口（默认SSL端口为6380，非SSL端口为6379，推荐使用SSL连接）
• 确认Spark集群已匹配对应版本的spark-redis连接器，Spark 3.x版本推荐使用com.redislabs:spark-redis_2.12:3.1.0，提交作业时可以通过--packages参数指定依赖，也可以提前将依赖包导入集群的依赖库。

代码实现步骤 #

1. 初始化SparkSession并配置Redis连接参数

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("WriteDataToAzureRedis") \
    .config("spark.redis.host", "你的Redis实例名.redis.cache.windows.net") \
    .config("spark.redis.port", "6380") \
    .config("spark.redis.auth", "你的Redis访问密钥") \
    .config("spark.redis.ssl", "true") \ # Azure Redis默认要求SSL连接，必须开启该配置
    .getOrCreate()

2. 适配键值列要求
   spark-redis默认会读取名为key的列作为Redis键，名为value的列作为Redis值，你可以将原DataFrame的两列重命名适配要求：

# 假设你的原始DataFrame为df，第一列为key列，第二列为value列
redis_write_df = df.selectExpr("原第一列名 as key", "原第二列名 as value")

如果你不想修改原DataFrame的列名，也可以在写入时直接指定键值列：

df.write \
    .format("org.apache.spark.sql.redis") \
    .option("key.column", "原第一列名") \
    .option("value.column", "原第二列名") \
    .mode("append") \
    .save()

3. 执行写入操作

redis_write_df.write \
    .format("org.apache.spark.sql.redis") \
    .option("batch.size", 1000) \ # 可选，配置每批写入的键数量，优化写入性能
    .option("ttl", 3600) \ # 可选，配置键的过期时间，单位为秒，不配置则为永久存储
    .mode("append") \ # 可替换为overwrite覆盖已存在的同键名数据
    .save()

注意事项 #

• 如果第二列是复杂结构化数据，spark-redis会自动将其序列化为JSON格式存储，读取时也支持自动反序列化
• 如果出现连接失败的问题，先检查Azure Redis的防火墙配置，是否放行了Spark集群的出站IP，同时确认SSL配置、端口、密钥填写正确

内容的提问来源于stack exchange，提问作者Ajay Ganti