Ошибка при создании таблицы Hive с использованием схемы dataframe в PySpark

Question

Я пытаюсь получить схему из фрейма данных PySpark и создать таблицу Hive, используя значения столбцов. Поскольку я делю таблицу, мне нужно создать таблицу Hive, а затем загрузить данные.

rawSchema = df.schema 
schemaString = rawSchema.fields.map(lambda field : field.name+" "+field.dataType.typeName).collect(Collectors.joining(","));      

sqlQuery = "CREATE TABLE IF NOT EXISTS Hive_table COLUMNS (" + schemaString + ") PARTITIONED BY (p int) STORED AS PARQUET LOCATION  Hive_table_path;"

Но при получении ошибки во 2-й строке:

AttributeError: у объекта 'list' нет атрибута 'map'

This Scala Код работает нормально, но мне нужно конвертировать его для PySpark

StructType my_schema = my_DF.schema();
String columns = Arrays.stream(my_schema.fields()).map(field ->field.name()+" "+field.dataType().typeName()).collect(Collectors.joining(","));

Пожалуйста, помогите!

Answer 1

Вы можете create list и затем join список для создания schemaString.

Example:

df.show()
#+---+----+
#| id|name|
#+---+----+
#|  a|   2|
#|  b|   3|
#+---+----+

df.schema
#StructType(List(StructField(id,StringType,true),StructField(name,StringType,true)))

schemaString=','.join([f.name+" "+f.dataType.typeName() for f in df.schema.fields])
#'id string,name string'

sqlQuery = "CREATE TABLE IF NOT EXISTS Hive_table COLUMNS (" + schemaString + ") PARTITIONED BY (p int) STORED AS PARQUET LOCATION  Hive_table_path;"

#'CREATE TABLE IF NOT EXISTS Hive_table COLUMNS (id string,name string) PARTITIONED BY (p int) STORED AS PARQUET LOCATION  Hive_table_path;'


#Using df.dtypes

schemaString=",".join([' '.join(w) for w in df.dtypes])
#'id string,name string'

sqlQuery = "CREATE TABLE IF NOT EXISTS Hive_table COLUMNS (" + schemaString + ") PARTITIONED BY (p int) STORED AS PARQUET LOCATION  Hive_table_path;"
#'CREATE TABLE IF NOT EXISTS Hive_table COLUMNS (id string,name string) PARTITIONED BY (p int) STORED AS PARQUET LOCATION  Hive_table_path;'