Как написать из PySpark DStream в Redis? - PullRequest
Новая
       34

Как написать из PySpark DStream в Redis?

0 голосов
/ 21 июня 2019

Я использую PySpark 2.3.1 для чтения потока значений из Kafka как DStream s. Я хочу выполнить некоторые преобразования этих данных, например взять скользящее среднее и сохранить его в Redis. Мой код искровой работы выглядит примерно так: 

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, и документация для Redis говорит, что из PySpark могут быть записаны только Dataframes. К счастью, 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 за пределами моей истории консоли, поэтому я не могу понять, что является причиной проблемы.

1 Ответ

0 голосов
/ 22 июня 2019

Вот пример Word Counter, который сохраняет результат в Redis: 

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"

Если вы хотитечтобы сохранить DataFrame с некоторыми значимыми именами столбцов, а не _1, _2 и т. д., вы можете переименовать столбцы следующим образом: 

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"

Надеюсь, это поможет!

...