Как переименовать все столбцы, привести типы данных на основе схемы / прочитать из файла CSV после создания кадра данных в pyspark

Question

Создан примерный фрейм данных в pyspark

   from pyspark.sql.types import *
   from pyspark.sql.types import StructField
   from pyspark.sql import types
   testdata = [("aaaa",1,50.0,"05-APR-2020"),
               ("bbbb",2,100.0,"06-APR-2020")]
   dataschema = types.StructType([
        types.StructField('col1', types.StringType(), True),
        types.StructField('col2', types.IntegerType(), True),
        types.StructField('col3', types.DoubleType(), True),
        types.StructField('col4', types.DateType(), True)
    ])
   testdf2 = spark.createDataFrame(
          spark.sparkContext.parallelize(testdata),
          dataschema
          )
   testdf2.printSchema()
   testdf2.show()

Получение следующей ошибки.

TypeError: поле col4: DateType не может принять объект '05 -APR-2020 'в типе

Если у меня есть список с 2 столбцами, один с новым столбцом, а другой с типом данных. Как я могу переименовать все столбцы и привести тип данных каждого столбца на основе списка или CSV / json file

Answer 1

По умолчанию Spark не преобразует строку в date type.

. Нам нужно использовать модуль datetime, чтобы определить тогда наши входные данные при чтении с помощью схемы spark создают col4 для типа даты.

Example:

import datetime
from pyspark.sql.types import *
from pyspark.sql.types import StructField
from pyspark.sql import types
testdata = [("aaaa",1,50.0,datetime.datetime.strptime('05-APR-2020','%d-%b-%Y')),
            ("bbbb",2,100.0,datetime.datetime.strptime('06-APR-2020','%d-%b-%Y'))]

dataschema = types.StructType([
        types.StructField('col1', types.StringType(), True),
        types.StructField('col2', types.IntegerType(), True),
        types.StructField('col3', types.DoubleType(), True),
        types.StructField('col4', types.DateType(), True)
    ])

testdf2 = spark.createDataFrame(
          spark.sparkContext.parallelize(testdata),
          dataschema
          )
testdf2.printSchema()
#root
# |-- col1: string (nullable = true)
# |-- col2: integer (nullable = true)
# |-- col3: double (nullable = true)
# |-- col4: date (nullable = true)


testdf2.show()
#+----+----+-----+----------+
#|col1|col2| col3|      col4|
#+----+----+-----+----------+
#|aaaa|   1| 50.0|2020-04-05|
#|bbbb|   2|100.0|2020-04-06|
#+----+----+-----+----------+

---

Другой способ заключается в определении stringtype для col4 с последующим преобразованием в date с использованием функции to_date.

dataschema = types.StructType([
        types.StructField('col1', types.StringType(), True),
        types.StructField('col2', types.IntegerType(), True),
        types.StructField('col3', types.DoubleType(), True),
        types.StructField('col4', types.StringType(), True)
    ])

testdata = [("aaaa",1,50.0,"05-APR-2020"),
               ("bbbb",2,100.0,"06-APR-2020")]

spark.createDataFrame(testdata,dataschema).withColumn("col4",to_date(col("col4"),"dd-MMM-yyyy")).printSchema()
#root
# |-- col1: string (nullable = true)
# |-- col2: integer (nullable = true)
# |-- col3: double (nullable = true)
# |-- col4: date (nullable = true)

spark.createDataFrame(testdata,dataschema).withColumn("col4",to_date(col("col4"),"dd-MMM-yyyy")).show()
#+----+----+-----+----------+
#|col1|col2| col3|      col4|
#+----+----+-----+----------+
#|aaaa|   1| 50.0|2020-04-05|
#|bbbb|   2|100.0|2020-04-06|
#+----+----+-----+----------+