Построение StructType из фрейма данных в pyspark

Я новая искра и питон и столкнулся с этой трудностью построения схемы из файла метаданных, который можно применить к моему файлу данных. Сценарий: файл метаданных для файла данных (формат csv) содержит столбцы и их типы: например:

id,int,10,"","",id,"","",TRUE,"",0
created_at,timestamp,"","","",created_at,"","",FALSE,"",0

Я успешно преобразовал это в dataframe, который выглядит так:

+--------------------+---------------+
|                name|           type|
+--------------------+---------------+
|                  id|  IntegerType()|
|          created_at|TimestampType()|
|          updated_at|   StringType()|

Но когда я пытаюсь преобразовать это в формат StructField, используя этот

fields = schemaLoansNew.map(lambda l:([StructField(l.name, l.type, 'true')]))

ИЛИ

schemaList = schemaLoansNew.map(lambda l: ("StructField(" + l.name + "," + l.type + ",true)")).collect()

И затем преобразуйте его в StructType, используя

schemaFinal = StructType(schemaList)

Я получаю следующую ошибку:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/opt/mapr/spark/spark-1.4.1/python/pyspark/sql/types.py", line 372, in __init__
assert all(isinstance(f, DataType) for f in fields), "fields should be a list of DataType"
AssertionError: fields should be a list of DataType

Я застрял на этом из-за недостатка знаний в Data Frames, не могли бы вы посоветовать, как действовать дальше. как только у меня будет готовая схема, я хочу использовать createDataFrame для применения к моему файлу данных. Этот процесс необходимо выполнить для многих таблиц, поэтому я не хочу жестко кодировать типы, а использовать файл метаданных для построения схемы, а затем применять к RDD.

Спасибо заранее.

Ответы

Ответ 1

Поля имеют аргумент, который должен быть списком объектов DataType. Это:

.map(lambda l:([StructField(l.name, l.type, 'true')]))

генерирует после collect a list of lists of tuples (Rows) DataType (list[list[tuple[DataType]]]), не говоря уже о том, что аргумент nullable должен быть логическим, а не строкой.

Вторая попытка:

.map(lambda l: ("StructField(" + l.name + "," + l.type + ",true)")).

генерирует после collect a list объектов str.

Правильная схема для показанной вами записи должна выглядеть примерно так:

from pyspark.sql.types import *

StructType([
    StructField("id", IntegerType(), True),
    StructField("created_at", TimestampType(), True),
    StructField("updated_at", StringType(), True)
])

Хотя использование распределенных структур данных для такой задачи является серьезным излишеством, не говоря уже о неэффективности, вы можете попытаться настроить свое первое решение следующим образом:

StructType([
    StructField(name, eval(type), True) for (name, type) in  df.rdd.collect()
])

но это не особенно безопасно (eval). Может быть проще построить схему из JSON/dictionary. Предполагая, что у вас есть функция, которая отображает из описания типа в имя канонического типа:

def get_type_name(s: str) -> str:
    """
    >>> get_type_name("int")
    'integer'
    """
    _map = {
        'int': IntegerType().typeName(),
        'timestamp': TimestampType().typeName(),
        # ...
    } 
    return _map.get(s, StringType().typeName())

Вы можете создать словарь следующей формы:

schema_dict = {'fields': [
    {'metadata': {}, 'name': 'id', 'nullable': True, 'type': 'integer'},
    {'metadata': {}, 'name': 'created_at', 'nullable': True, 'type': 'timestamp'}
], 'type': 'struct'}

и подайте его на StructType.fromJson:

StructType.fromJson(schema_dict)

Ответ 2

Ниже приведены шаги для изменения объектов типа данных.

data_schema=[
    StructField("age", IntegerType(), True),
    StructField("name", StringType(), True)
]



final_struct=StructType(fields=data_schema)

df=spark.read.json('/home/abcde/Python-and-Spark-for-Big-Data-master/Spark_DataFrames/people.json', schema=final_struct)



df.printSchema()

root
 |-- age: integer (nullable = true)
 |-- name: string (nullable = true)

Ответ 3

val columns: Array[String] = df1.columns
val reorderedColumnNames: Array[String] = df2.columns //or do the reordering you want
val result: DataFrame = dataFrame.select(reorderedColumnNames.head, reorderedColumnNames.tail: _*)