Построить схему зажигания из схемы JSON

Question

Я пытаюсь создать схему искры, которая нужна для расширения возможностей при создании кадра данных. Я могу сгенерировать схему 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}

, но когда тип является структурой, он создает список словаря всех элементовструктурируйте и присвойте его типу и делайте это рекурсивно, пока он не найдет какую-либо структуру.

Все, что я могу собрать, это

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

Наконец-то удалось решить

def struct_definition(column_name, column_type):
    column_dict = {"metadata": {}, "name": column_name, "nullable": True, "type": column_type}
    return column_dict


def convert_to_json_array(struct_def):
    striped = struct_def.lstrip('struct')
    striped = striped.lstrip('<')
    striped = striped.rstrip('>')
    main_struct_list = []
    if striped.__contains__('struct'):
        name = striped.split(':')[0]
        json = {'Name': name, 'Type': striped.lstrip(name + ':') + '>'}
        main_struct_list.append(json)
    else:
        for i in striped.split(','):
            key_value = i.split(':')
            normalized_json = {'Name': key_value[0], 'Type': key_value[1]}
            main_struct_list.append(normalized_json)
    return main_struct_list


def to_json(input_list):
    main_struct_list = []
    for x in input_list:
        column_name = x['Name']
        column_type = x['Type']
        if column_type.startswith('struct'):
            main_struct_list.append(
                struct_definition(column_name,
                                  {'fields': to_json(convert_to_json_array(column_type)), 'type': 'struct'}))
        else:
            main_struct_list.append(struct_definition(column_name, column_type))
    return main_struct_list


if __name__ == '__main__':
    sample_list = [{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'}]
    main_struct_dict = {'fields': to_json(sample_list), 'type': 'struct'}
    print(main_struct_dict)

Answer 2

Почему бы не использовать стандартную json-схему https://json -schema.org / понимание-json-schema / about.html и использовать эту json-схему для преобразования в схему spark https://github.com/zalando-incubator/spark-json-schema#quickstart

Answer 3

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

def extract_struct(text):
    stop = 7
    flag = 1
    for c in text[7:]:
        stop += 1
        if c == "<":
            flag += 1
        if c == ">":
            flag -= 1
        if flag == 0:
            return text[:stop], text[stop:]

. Это вернет вам извлеченную структуру и оставшийся текст из определения столбца.Например,

extract_struct("struct<a:int,b:double>,c:string")

вернет

("struct<a:int,d:double>", "c:string").

Во-вторых, нам нужно пройтись по каждому типу столбца и получить определение элементов структуры:

def parse(s, node):
    while s != '':
        # Strip column name
        col_name = s.partition(':')[0]
        s = s.partition(':')[2]

        # If column type is struct, parse it as well
        if s.startswith('struct'):
            col_type, s = extract_struct(s)
            node[col_name] = {}
            parse(col_type[7:-1], node[col_name])
        else:
            # Just add column definition
            col_type = s.partition(',')[0]
            node[col_name] = {
                "metadata": {},
                "name": col_name,
                "nullable": True,
                "type": col_type
            }

        # Go to next entry
        s = s.partition(',')[2]

Вышеприведенный метод просто добавляет новый столбец к текущему узлу в дереве схемы, если тип столбца простой, в противном случае он извлекает имя и структуру и рекурсивно просматривает дочерние записи структуры.Теперь нам осталось только просмотреть каждую колонку и проанализировать их.Итак, после обобщения вышеописанного одним способом:

def build(columns):
    def extract_struct(text):
        stop = 7
        flag = 1
        for c in text[7:]:
            stop += 1
            if c == '<':
                flag += 1
            if c == '>':
                flag -= 1
            if flag == 0:
                return text[:stop], text[stop:]

    def parse(s, node):
        while s != '':
            # Strip column name
            col_name = s.partition(':')[0]
            s = s.partition(':')[2]

            # If column type is struct, parse it as well
            if s.startswith('struct'):
                col_type, s = extract_struct(s)
                node[col_name] = {}
                parse(col_type[7:-1], node[col_name])
            else:
                # Just add column definition
                col_type = s.partition(',')[0]
                node[col_name] = {
                    "metadata": {},
                    "name": col_name,
                    "nullable": True,
                    "type": col_type
                }

            # Go to next entry
            s = s.partition(',')[2]

    schema = {}
    for column in columns:
        parse("{}:{}".format(column['Name'], column['Type']), schema)
    return schema

Теперь, если вы запустите его в своем списке примеров, вы получите следующий словарь (который легко преобразовать в список столбцов, но порядок нев любом случае это не имеет значения):

{
  "cloud_events_version": {
    "nullable": true, 
    "type": "string", 
    "name": "cloud_events_version", 
    "metadata": {}
  }, 
  "event_type": {
    "nullable": true, 
    "type": "string", 
    "name": "event_type", 
    "metadata": {}
  }, 
  "event_time": {
    "nullable": true, 
    "type": "string", 
    "name": "event_time", 
    "metadata": {}
  }, 
  "event_validated_ts": {
    "nullable": true, 
    "type": "string", 
    "name": "event_validated_ts", 
    "metadata": {}
  }, 
  "event_type_version": {
    "nullable": true, 
    "type": "string", 
    "name": "event_type_version", 
    "metadata": {}
  }, 
  "source": {
    "nullable": true, 
    "type": "string", 
    "name": "source", 
    "metadata": {}
  }, 
  "extensions": {
    "client_common": {
      "adobe_sdid": {
        "nullable": true, 
        "type": "string", 
        "name": "adobe_sdid", 
        "metadata": {}
      }, 
      "auth_state": {
        "nullable": true, 
        "type": "string", 
        "name": "auth_state", 
        "metadata": {}
      }, 
      "client_ip_address": {
        "nullable": true, 
        "type": "string", 
        "name": "client_ip_address", 
        "metadata": {}
      }, 
      "url": {
        "nullable": true, 
        "type": "string", 
        "name": "url", 
        "metadata": {}
      }, 
      "client_experiments": {
        "nullable": true, 
        "type": "string", 
        "name": "client_experiments", 
        "metadata": {}
      }, 
      "referral_url": {
        "nullable": true, 
        "type": "string", 
        "name": "referral_url", 
        "metadata": {}
      }, 
      "page_name": {
        "nullable": true, 
        "type": "string", 
        "name": "page_name", 
        "metadata": {}
      }, 
      "user_agent": {
        "nullable": true, 
        "type": "string", 
        "name": "user_agent", 
        "metadata": {}
      }, 
      "uvn": {
        "nullable": true, 
        "type": "string", 
        "name": "uvn", 
        "metadata": {}
      }, 
      "chegg_uuid": {
        "nullable": true, 
        "type": "string", 
        "name": "chegg_uuid", 
        "metadata": {}
      }, 
      "adobe_mcid": {
        "nullable": true, 
        "type": "string", 
        "name": "adobe_mcid", 
        "metadata": {}
      }, 
      "device_id": {
        "nullable": true, 
        "type": "string", 
        "name": "device_id", 
        "metadata": {}
      }
    }
  }, 
  "validated_message": {
    "nullable": true, 
    "type": "string", 
    "name": "validated_message", 
    "metadata": {}
  }, 
  "event_validation_status": {
    "nullable": true, 
    "type": "string", 
    "name": "event_validation_status", 
    "metadata": {}
  }, 
  "data": {
    "school_type": {
      "nullable": true, 
      "type": "string", 
      "name": "school_type", 
      "metadata": {}
    }, 
    "reg_source_product": {
      "nullable": true, 
      "type": "string", 
      "name": "reg_source_product", 
      "metadata": {}
    }, 
    "member_role": {
      "nullable": true, 
      "type": "string", 
      "name": "member_role", 
      "metadata": {}
    }, 
    "email": {
      "nullable": true, 
      "type": "string", 
      "name": "email", 
      "metadata": {}
    }, 
    "year_in_college": {
      "nullable": true, 
      "type": "int", 
      "name": "year_in_college", 
      "metadata": {}
    }
  }
}

Наконец, обратите внимание, что это работает только для простых типов и struct (не для array или map типов).Но его довольно легко распространить и на другие сложные типы.