python - 如何从 PySpark DStream 写入 Redis？

Question

我正在使用 PySpark 2.3.1 从 Kafka 读取值流作为 DStream。我想对这些数据做一些转换，比如取移动平均，并将其保存到 Redis。我的 spark 作业代码看起来有点像这样:

batch_duration = 1

# Initialize session
spark_session = SparkSession \
    .builder \
    .appName("my-app") \
    .getOrCreate()

spark_context = spark_session.sparkContext

# Create streaming context (=connection to Spark)
streaming_context = StreamingContext(spark_context, batch_duration)

# Read from Kafka
input = KafkaUtils \
    .createDirectStream(streaming_context, ['price'], {"metadata.broker.list": kafka_urls})

然后我可以用这样的行来转换它:

jsons = input.window(5000).map(lambda t: t[1]).map(json.loads)
prices = jsons.map(lambda d: d['price'])
total = prices.reduce(lambda x, y: x + y)

但是 total 在这种情况下仍然是一个 DStream，并且 documentation for Redis说只能从 PySpark 写入数据帧。幸运的是，DStream 在运行时会生成周期性的 RDD - 所以我必须弄清楚如何将 RDD 转换为 Dataframe。

我试过了

total.foreachRDD(lambda rdd:
                 rdd.toDF().write.format("org.apache.spark.sql.redis") \
                 .option("table", "people") \
                 .option("key.column", "name") \
                 .save())

不可否认，这是从网络上的其他地方盲目复制和粘贴的，因此 option 调用几乎可以肯定与我的数据模式不匹配。我希望破译异常并弄清楚下一步要去哪里。不幸的是，在我的 Spark 集群上运行它会打印多行 Java 堆栈跟踪，并将原始 Python 异常滚动到我的控制台历史记录之外，所以我无法找出导致问题的原因。

Answer 1

这是一个将结果保存到 Redis 的 Word Counter 示例:

import sys

from pyspark.sql import SparkSession
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from pyspark.sql import SQLContext

def save_rdd(rdd):
    if not rdd.isEmpty():
        df = rdd.toDF()
        df.show()
        df.write.format("org.apache.spark.sql.redis").option("table", "counts").option("key.column", "_1").save(mode='append')

if __name__ == "__main__":
    spark = SparkSession \
        .builder \
        .appName("Example") \
        .getOrCreate()

    sc = spark.sparkContext
    sc.setLogLevel("ERROR")
    ssc = StreamingContext(sc, 2)

    brokers, topic = sys.argv[1:]
    kvs = KafkaUtils.createDirectStream(ssc, [topic], {"metadata.broker.list": brokers})
    lines = kvs.map(lambda x: x[1])
    counts = lines.flatMap(lambda line: line.split(" ")) \
        .map(lambda word: (word, 1)) \
        .reduceByKey(lambda a, b: a + b)

    counts.foreachRDD(save_rdd)

    ssc.start()
    ssc.awaitTermination()

提交命令:

./bin/spark-submit --master spark://Oleksiis-MacBook-Pro.local:7077 --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.0,com.redislabs:spark-redis:2.4.0 ~/Projects/spark-redis-test/src/main/scala/com/redislabs/provider/test/spark-direct-kafka.py localhost:9092 new_topic

请注意，我包含了 com.redislabs:spark-redis:2.4.0 包。

写一些词到 new_topic:

./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic new_topic

>a b c a
>a b b

输出应该在 Redis 中显示为哈希值，其中键对应于输入词:

./redis-cli

keys counts:*

1) "counts:a"
2) "counts:b"
3) "counts:c"

127.0.0.1:6379> hgetall counts:a
1) "_2"
2) "2"

如果您想使用一些有意义的列名而不是 _1、_2 等来保存 DataFrame - 您可以像这样重命名列:

from pyspark.sql.functions import col

def save_rdd(rdd):
    if not rdd.isEmpty():
        df = rdd.toDF().select(col("_1").alias("word"), col("_2").alias("count"))
        df.show()
        df.write.format("org.apache.spark.sql.redis").option("table", "counts").option("key.column","word").save(mode='append')

请注意，现在我们将 key.coumn 参数设置为 word。

现在Redis中的字段名是“count”:

127.0.0.1:6379> hgetall counts:abc
1) "count"
2) "1"

希望对您有所帮助!