从json模式构建spark模式

Question

我正在尝试构建一个想要显式提供的spark模式，同时创建数据框我可以使用下面生成json模式

from pyspark.sql.types import StructType
# Save schema from the original DataFrame into json:
schema_json = df.schema.json()

这给了我

{"fields":[{"metadata":{},"name":"cloud_events_version","nullable":true,"type":"string"},{"metadata":{},"name":"data","nullable":true,"type":{"fields":[{"metadata":{},"name":"email","nullable":true,"type":"string"},{"metadata":{},"name":"member_role","nullable":true,"type":"string"},{"metadata":{},"name":"reg_source_product","nullable":true,"type":"string"},{"metadata":{},"name":"school_type","nullable":true,"type":"string"},{"metadata":{},"name":"year_in_college","nullable":true,"type":"long"}],"type":"struct"}},{"metadata":{},"name":"event_time","nullable":true,"type":"string"},{"metadata":{},"name":"event_type","nullable":true,"type":"string"},{"metadata":{},"name":"event_type_version","nullable":true,"type":"string"},{"metadata":{},"name":"event_validated_ts","nullable":true,"type":"string"},{"metadata":{},"name":"event_validation_status","nullable":true,"type":"string"},{"metadata":{},"name":"extensions","nullable":true,"type":{"fields":[{"metadata":{},"name":"client_common","nullable":true,"type":{"fields":[{"metadata":{},"name":"adobe_mcid","nullable":true,"type":"string"},{"metadata":{},"name":"adobe_sdid","nullable":true,"type":"string"},{"metadata":{},"name":"auth_state","nullable":true,"type":"string"},{"metadata":{},"name":"uuid","nullable":true,"type":"string"},{"metadata":{},"name":"client_experiments","nullable":true,"type":"string"},{"metadata":{},"name":"client_ip_address","nullable":true,"type":"string"},{"metadata":{},"name":"device_id","nullable":true,"type":"string"},{"metadata":{},"name":"page_name","nullable":true,"type":"string"},{"metadata":{},"name":"referral_url","nullable":true,"type":"string"},{"metadata":{},"name":"url","nullable":true,"type":"string"},{"metadata":{},"name":"user_agent","nullable":true,"type":"string"},{"metadata":{},"name":"uvn","nullable":true,"type":"string"}],"type":"struct"}}],"type":"struct"}},{"metadata":{},"name":"source","nullable":true,"type":"string"},{"metadata":{},"name":"validated_message","nullable":true,"type":"string"},{"metadata":{},"name":"year","nullable":true,"type":"integer"},{"metadata":{},"name":"mon","nullable":true,"type":"integer"},{"metadata":{},"name":"day","nullable":true,"type":"integer"},{"metadata":{},"name":"hour","nullable":true,"type":"integer"}],"type":"struct"}

但为此，我需要解析数据帧，这需要一些时间，我试图避免

我能做的一件事是从我们内部的目录中获取所需的模式。这给了类似的东西

[{u'Name': u'cloud_events_version', u'Type': u'string'},
 {u'Name': u'event_type', u'Type': u'string'},
 {u'Name': u'event_time', u'Type': u'string'},
 {u'Name': u'data', u'Type': u'struct<school_type:string,reg_source_product:string,member_role:string,email:string,year_in_college:int>'},
 {u'Name': u'source', u'Type': u'string'},
 {u'Name': u'extensions', u'Type': u'struct<client_common:struct<auth_state:string,client_ip_address:string,client_experiments:string,uvn:string,device_id:string,adobe_sdid:string,url:string,page_name:string,user_agent:string,uuid:string,adobe_mcid:string,referral_url:string>>'},
 {u'Name': u'event_type_version', u'Type': u'string'},
 {u'Name': u'event_validation_status', u'Type': u'string'},
 {u'Name': u'event_validated_ts', u'Type': u'string'},
 {u'Name': u'validated_message', u'Type': u'string'}]

我正在尝试编写一个递归python查询，生成上面的json。逻辑是循环遍历此dict列表，并在类型为字符串时为此字典指定名称和类型

{"metadata" : {},"name" : columnName,"nullable" : True,"type" : columnType}

但是当类型是struct时，它会创建struct的所有元素的字典列表，并将其分配给type并以递归方式执行，直到找不到任何struct。

我只能鼓起勇气

def structRecursive(columnName,columnType):
    if "struct" not in columnType:
        ColumnDict = {"metadata" : {},"name" : columnName,"nullable" : True,"type" : columnType}
    else:
        structColumnList = []
        structColumnDict = {
            'metadata': {},
            'name': columnName,
            'nullable': True,
            'type': {'fields': structColumnList, 'type': 'struct'}
        }
        if columnType.count('struct<')==1:
            structCol = columnName
            structColList = columnType.encode('utf-8').replace('struct<',
                    '').replace('>', '').split(',')
            for item in structColList:
                fieldName = item.split(':')[0]
                dataType = item.split(':')[1]
                nodeDict = {}
                nodeDict['metadata'] = {}
                nodeDict['name'] = '{}'.format(fieldName)
                nodeDict['nullable'] = True
                nodeDict['type'] = '{}'.format(dataType)
                structColumnList.append(nodeDict)
        else:
            columnName = columnType.replace('struct<','',1).replace('>','').split(':')[0]
            columnType = columnType.split("{}:".format(columnName),1)[1].replace('>','',1)
        return structColumnDict

MainStructList = []
MainStructDict = {'fields': MainStructList, 'type': 'struct'}
for item in ListOfDict :
    columnName = item['Name'].encode('utf-8')
    columnType = item['Type'].encode('utf-8')
    MainStructList.append(structRecursive(columnName,columnType))

当然，这并没有给出理想的结果。很想在这里得到一些建议。

Answer 1

另一答案

Answer 2

另一答案