CODEWITHSUNDEEP
pythonapache-sparkpyspark
Ivan
Ivan

Reputation: 20101

VectorType for StructType in Pyspark Schema

I'm reading a parquet file that has the following schema:

df.printSchema()

root
 |-- time: integer (nullable = true)
 |-- amountRange: integer (nullable = true)
 |-- label: integer (nullable = true)
 |-- pcaVector: vector (nullable = true)

Now I want to test Pyspark structured streaming and I want to use the same parquet files. The closest schema that I was able to create was using ArrayType, but it doesn't work:

schema = StructType(
    [
        StructField('time', IntegerType()),
        StructField('amountRange', IntegerType()),
        StructField('label', IntegerType()),
        StructField('pcaVector', ArrayType(FloatType()))

    ]
)
df_stream = spark.readStream\
    .format("parquet")\
    .schema(schema)\
    .load("/home/user/test_arch/data/fraud/")

Caused by: java.lang.ClassCastException: Expected instance of group converter but got "org.apache.spark.sql.execution.datasources.parquet.ParquetPrimitiveConverter"
        at org.apache.parquet.io.api.Converter.asGroupConverter(Converter.java:37)
        at org.apache.spark.sql.execution.datasources.parquet.ParquetRowConverter$RepeatedGroupConverter.<init>(ParquetRowConverter.scala:659)
        at org.apache.spark.sql.execution.datasources.parquet.ParquetRowConverter.org$apache$spark$sql$execution$datasources$parquet$ParquetRowConverter$$newConverter(ParquetRowConverter.scala:308)

How can I create a schema with VectorType, that seems to exist only for Scala, for the StructType in Pyspark?

Upvotes: 7

Views: 4364

Answers (1)

user10135807
user10135807

Reputation: 96

The type is VectorUDT

from pyspark.ml.linalg import VectorUDT

StructField('pcaVector', VectorUDT())

Upvotes: 8

Related Questions