Как преобразовать строку rdd в фрейм данных с помощью структуры json в pyspark? - PullRequest
Новая
       33

Как преобразовать строку rdd в фрейм данных с помощью структуры json в pyspark?

1 голос
/ 16 марта 2019

Я отправляю следующий файл json по пути "/ home / host / test", чтобы программа могла перехватить его с помощью потоковой передачи с плавающей точкой и выполнить запросы о нем. 

{"id": "1", description: "test"}
{"id": "1", description: "test"}

Но когда я выполняю запрос, это выглядит как следующая структура 

root
   | --word: String (Nulleable = true)

и я получаю следующий результат: 

+ ------------------- +
| word |
---------------------
| {"id": "1", "test"}
| {"id": "1", "test"}

Мне нужно, чтобы структура выглядела так 

root
   | --id: String (Nulleable = true)
   | --description string (Nulleable = true)

и мне нужно получить результат, подобный следующему 

 ----------------
| id | description
----------------
| "1" | "test" |
| "1" | "test" |
----------------

это мой код pyspkark 

from __future__ import print_function
import os
import sys
from pyspark import SparkContext
from pyspark.sql.functions import col, explode
from pyspark.streaming import StreamingContext
from pyspark.sql import SQLContext, Row
from pyspark.sql import SQLContext


if __name__ == "__main__":

sc = SparkContext(appName="PythonSqlNetworkWordCount")
ssc = StreamingContext(sc, 3)
sqlcontextoriginal = SQLContext(sc)

# Create a socket stream on target ip:port and count the
# words in input stream of \n delimited text (eg. generated by 'nc')
lines = ssc.textFileStream("/home/host/test")

# Convert RDDs of the words DStream to DataFrame and run SQL query
def process(time, rdd):
    print("========= %s =========" % str(time))

    try:
        # Get the singleton instance of SQLContext
        sqlContext = SQLContext(rdd.context)
        # Convert RDD[String] to RDD[Row] to DataFrame
        rowRdd = rdd.map(lambda w: Row(word=w))
        wordsDataFrame = sqlContext.createDataFrame(rowRdd).toJSON()

        json = sqlContext.read.json(wordsDataFrame)
        # Register as table
        json.createOrReplaceTempView("words")
        json.printSchema()

        wordCountsDataFrame = sqlContext.sql("select * from words ")
        wordCountsDataFrame.show()

    except:
        pass

lines.foreachRDD(process)
ssc.start()
ssc.awaitTermination()

1 Ответ

0 голосов
/ 16 марта 2019

Хорошо, я нашел решение.

Мне пришлось использовать sql.read.json, передавая его непосредственно в качестве параметра rdd.

json = sqlContext.read.json (rdd)

...