Создать фрейм данных со схемой, предоставленной в виде файла JSON - PullRequest
Новая
       28

Создать фрейм данных со схемой, предоставленной в виде файла JSON

1 голос
/ 06 марта 2020

Как я могу создать фрейм данных pyspark с 2 JSON файлами?

  • file1: этот файл содержит полные данные
  • file2: этот файл содержит только схему данных file1.

file1 

{"RESIDENCY":"AUS","EFFDT":"01-01-1900","EFF_STATUS":"A","DESCR":"Australian Resident","DESCRSHORT":"Australian"}

file2 

[{"fields":[{"metadata":{},"name":"RESIDENCY","nullable":true,"type":"string"},{"metadata":{},"name":"EFFDT","nullable":true,"type":"string"},{"metadata":{},"name":"EFF_STATUS","nullable":true,"type":"string"},{"metadata":{},"name":"DESCR","nullable":true,"type":"string"},{"metadata":{},"name":"DESCRSHORT","nullable":true,"type":"string"}],"type":"struct"}]

Ответы [ 2 ]

2 голосов
/ 07 марта 2020

Сначала вы должны прочитать файл схемы, используя Python json.load, затем преобразовать его в DataType, используя StructType.fromJson. 

import json
from pyspark.sql.types import StructType

with open("/path/to/file2.json") as f:
    json_schema = json.load(f)

schema = StructType.fromJson(json_schema[0])

Теперь просто передать эту схему в DataFrame Reader: 

df = spark.read.schema(schema).json("/path/to/file1.json")

df.show()

#+---------+----------+----------+-------------------+----------+
#|RESIDENCY|     EFFDT|EFF_STATUS|              DESCR|DESCRSHORT|
#+---------+----------+----------+-------------------+----------+
#|      AUS|01-01-1900|         A|Australian Resident|Australian|
#+---------+----------+----------+-------------------+----------+

РЕДАКТИРОВАТЬ:

Если файл, содержащий схему, находится в GCS, вы можете использовать Spark или Had oop API для получения содержимого файла. Вот пример использования Spark: 

file_content = spark.read.text("/path/to/file2.json").rdd.map(
    lambda r: " ".join([str(elt) for elt in r])
).reduce(
    lambda x, y: "\n".join([x, y])
)

json_schema = json.loads(file_content)
0 голосов
/ 11 марта 2020

Я нашел пакеты GCSFS для доступа к файлам в GCP Buckets: 

pip install gcsfs

import gcsfs

fs = gcsfs.GCSFileSystem(project='your GCP project name')

with fs.open('path/toread/sample.json', 'rb') as f:
     json_schema=json.load(f)
...