Обработка данных из потока kafka с помощью Pyspark - PullRequest
Новая
       118

Обработка данных из потока kafka с помощью Pyspark

1 голос
/ 18 июня 2020

Как выглядит консоль потребителя kafka: 

["2017-12-31 16:06:01", 12472391, 1]
["2017-12-31 16:06:01", 12472097, 1]
["2017-12-31 16:05:59", 12471979, 1]
["2017-12-31 16:05:59", 12472099, 0]
["2017-12-31 16:05:59", 12472054, 0]
["2017-12-31 16:06:00", 12472318, 0]
["2017-12-31 16:06:00", 12471979, 0]

Я хочу использовать pyspark для получения каждого значения в списке или df этих значений через заданный период.

Что я пробовал: 

sc = SparkContext(appName='PythonStreamingDirectKafka')
sc.setLogLevel("WARN")
spark = SparkSession(sc)
ssc = StreamingContext(sc, 10)

brokers, topic = sys.argv[1:]


kvs = KafkaUtils.createDirectStream(ssc, [topic],
    {'metadata.broker.list': brokers})


lines = kvs.map(lambda x: x[1])
text =  lines.flatMap(lambda line: line.split(" ")).pprint()

ssc.start()
ssc.awaitTermination()

текстовая переменная выше является объектом Dstream, и я не могу понять, как управлять им или преобразовывать его. Пройдено через множество блогов и документов.

Я хочу извлечь информацию в список python или pandas df, чтобы я мог с ней манипулировать

Был бы очень признателен за любую помощь. Спасибо ~

...