我正在使用 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 异常滚动到我的控制台历史记录之外,所以我无法找出导致问题的原因。
最佳答案
这是一个将结果保存到 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"
希望对您有所帮助!
关于python - 如何从 PySpark DStream 写入 Redis?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/56707978/