python - 从 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()

然后将其转换为 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

由于我对数据框缺乏了解，我被困在这个问题上，请问如何继续。准备好模式后，我想使用 createDataFrame 来应用于我的数据文件。必须对许多表完成此过程，因此我不想对类型进行硬编码，而是使用元数据文件构建模式，然后应用于 RDD。

提前致谢。

Answer 1

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

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

在 collect 之后生成一个list的 lists的 tuples ( Rows )属于 DataType ( list[list[tuple[DataType]]] ) 更不用说 nullable参数应该是 bool 值而不是字符串。

你的第二次尝试:

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

在 collect 之后生成一个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/字典构建模式可能更容易。假设您有从类型描述映射到规范类型名称的函数:

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)