Ошибка Pyspark DataframeType a: DoubleType не может принять объект типа a

Question

У меня есть эта функция

customSchema = StructType([ \
    StructField("a", Doubletype(), True), \
    StructField("b", Doubletype(), True),
    StructField("c", Doubletype(), True), 
    StructField("d", Doubletype(), True)])


n_1= sc.textFile("/path/*.txt")\
        .mapPartitions(lambda partition: csv.reader([line.replace('\0','') for line in partition],delimiter=';', quotechar='"')).filter(lambda line: len(line) > 1 )\
        .toDF(customSchema)

, которая создаст Dataframe, проблема в том, что '.mapPartitions' будет использовать как тип по умолчанию и мне нужно преобразовать его в DoubleType, прежде чем преобразовать в Dataframe. Есть идеи?

Примеры данных

[['0,01', '344,01', '0,00', '0,00']]

или просто работайте с

n_1= sc.textFile("/path/*.txt")\
        .mapPartitions(lambda partition: csv.reader([line.replace('\0','') for line in partition],delimiter=';', quotechar='"')).filter(lambda line: len(line) > 1 )\
       

Answer 1

Во-первых, нужно было собрать все элементы и создать матрицу (список списков), используя второй вариант.

n_1= sc.textFile("/path/*.txt")\
        .mapPartitions(lambda partition: csv.reader([line.replace('\0','') for line in partition],delimiter=';', quotechar='"')).filter(lambda line: len(line) > 1 )\
 

matrix  = n_1.collect()

Как только у нас есть это, необходимо знать, какой тип данных попадает в подсписки (в моем случае это была 'str').

matrix  =[[x.replace(',', '.') for x in i] for i in matrix ] # replace ',' for '.' in order to perform the data type convertion

matrix  = [[float(str(x)) for x in i] for i in matrix  ] #convert every sublist element into float

df = sc.parallelize(matrix).toDF()