课程 9：PySpark 数据挖掘及 Redis 结果存储

from pyspark.sql import SparkSession

# 创建一个 SparkSession
spark = SparkSession.builder \
    .appName("PySparkWordCountToRedis") \
    .master("local[*]") \
    .getOrCreate()

sc = spark.sparkContext # 获取 SparkContext 用于 RDD 操作

# 样本文本数据
text_data = [
    "Hello Spark, this is a sample text.",
    "Spark is great for big data processing.",
    "This sample text will be used for word count.",
    "Hello again Spark, process this text."
]

# 从字符串列表创建 RDD
text_rdd = sc.parallelize(text_data)
print("初始 RDD 内容:") # "Initial RDD content:"
for line in text_rdd.collect():
    print(line)

# FlatMap 将行分割成单词，转换为小写，并移除标点符号 (基本处理)
words_rdd = text_rdd.flatMap(lambda line: line.lower().replace('.', '').replace(',', '').split(" "))

# 将每个单词映射为 (单词, 1) 元组
word_map_rdd = words_rdd.map(lambda word: (word, 1))

# 过滤掉可能由多个空格产生的空字符串
word_map_rdd = word_map_rdd.filter(lambda x: x[0] != "")

# 按键聚合以统计每个单词的出现次数
word_counts_rdd = word_map_rdd.reduceByKey(lambda a, b: a + b)

# 行动：收集结果
word_counts_result = word_counts_rdd.collect()

print("\\nPySpark 词频统计结果:") # "Word Count Results from PySpark:"
for word, count in word_counts_result:
    print(f"'{word}': {count}")

import redis

# 假设 word_counts_result 可从上面的 PySpark 部分获得
# word_counts_result = [('hello', 2), ('spark', 3), ...] 

# Redis 连接详细信息
redis_host = 'localhost'
redis_port = 6379
redis_db = 0
redis_hash_key = 'pyspark_word_counts' # Redis 中哈希的名称

try:
    r = redis.Redis(host=redis_host, port=redis_port, db=redis_db, decode_responses=True)
    r.ping()
    print("\\n成功连接到 Redis！") # "Successfully connected to Redis!"

    # 为此示例清除哈希中的任何旧数据
    r.delete(redis_hash_key)

    # 将结果存储在 Redis 哈希中
    # 使用管道进行更高效的批量插入
    pipe = r.pipeline()
    for word, count in word_counts_result:
        if word: # 确保单词不为空
            pipe.hset(redis_hash_key, word, count)
    pipe.execute()
    
    print(f"已在 Redis 哈希 '{redis_hash_key}' 中存储 {len(word_counts_result)} 个词频统计。") # "Stored ... word counts in Redis hash ..."

except redis.exceptions.ConnectionError as e:
    print(f"无法连接到 Redis: {e}") # "Could not connect to Redis:"
    # 在实际应用程序中，应适当处理此错误
    # 对于此脚本，如果 Redis 不可用，我们将退出
    spark.stop() # 如果 Redis 失败，则停止 Spark 会话
    exit()
except Exception as e:
    print(f"Redis 操作期间发生错误: {e}") # "An error occurred during Redis operations:"
    spark.stop()
    exit()

try:
    # 从哈希中检索所有词频统计
    retrieved_word_counts = r.hgetall(redis_hash_key)
    print(f"\\n从 Redis 哈希 '{redis_hash_key}' 检索到的所有词频统计:") # "Retrieved all word counts from Redis hash..."
    if retrieved_word_counts:
        for word, count in retrieved_word_counts.items():
            print(f"'{word}': {count}")
    else:
        print("在 Redis 中未找到词频统计（或键不存在）。") # "No word counts found in Redis (or key does not exist)."

    # 检索特定单词的计数
    specific_word = 'spark'
    count_specific_word = r.hget(redis_hash_key, specific_word)
    if count_specific_word:
        print(f"\\n单词 '{specific_word}' 的计数: {count_specific_word}") # "Count for word..."
    else:
        print(f"\\n在 Redis 哈希中未找到单词 '{specific_word}'。") # "Word ... not found in Redis hash."

except Exception as e:
    print(f"从 Redis 检索数据时发生错误: {e}") # "An error occurred while retrieving data from Redis:"

finally:
    # 在所有操作结束时停止 SparkSession
    spark.stop()
    print("\\nSparkSession 已停止。") # "SparkSession stopped."

# docker-compose-spark-redis.yml
version: '3.8'

services:
  spark-master:
    image: bitnami/spark:3 # 使用特定版本，例如 3.2.1
    container_name: spark-master
    hostname: spark-master
    ports:
      - "8080:8080" # Spark Master Web UI
      - "7077:7077" # Spark Master RPC
    environment:
      - SPARK_MODE=master
      - SPARK_RPC_AUTHENTICATION_ENABLED=no
      - SPARK_RPC_ENCRYPTION_ENABLED=no
    networks:
      - spark_redis_network

  spark-worker:
    image: bitnami/spark:3
    container_name: spark-worker-1
    hostname: spark-worker-1
    depends_on:
      - spark-master
    ports:
      - "8081:8081" # Spark Worker Web UI
    environment:
      - SPARK_MODE=worker
      - SPARK_MASTER_URL=spark://spark-master:7077
      - SPARK_WORKER_MEMORY=1G
      - SPARK_WORKER_CORES=1
      - SPARK_RPC_AUTHENTICATION_ENABLED=no
      - SPARK_RPC_ENCRYPTION_ENABLED=no
    networks:
      - spark_redis_network

  redis:
    image: redis:alpine
    container_name: redis_service
    hostname: redis_service
    ports:
      - "6379:6379"
    networks:
      - spark_redis_network

networks:
  spark_redis_network:
    driver: bridge