在pyspark中从数据帧构建StructType

Question

我是新的spark和python，面临着从元数据文件构建模式的困难，该模式可以应用于我的数据文件。场景：数据文件的元数据文件（csv格式），包含列及其类型：例如：

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

我已成功将其转换为如下数据帧：

+--------------------+---------------+
|                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()

然后使用

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

一旦我准备好了模式，我想使用createDataFrame来应用于我的数据文件。这个过程必须为许多表完成，所以我不想硬编码类型，而是使用元数据文件来构建模式，然后应用于RDD。

提前感谢。

Answer 1

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: _*)

Answer 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)

Answer 3

字段的参数必须是DataType对象的列表。这：

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

在收集数据类型（list[list[tuple[DataType]]）的数据类型（list[tuple[DataType]]）的元组（Rows）的列表（list[tuple[DataType]]）的元组（Rows）的列表（list[tuple[DataType]]））的列表（）后生成，更不用说可为null的参数应该是布尔值而不是字符串。

您的第二次尝试：

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

在收集对象后生成对象的列表。

所显示记录的正确架构应大致如下所示：

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)