Here's my code
orion_apps_schema = StructType([
StructField("prot_category",IntegerType(),True),
StructField("prot_type",IntegerType(),True),
StructField("subs_no",LongType(),True),
StructField("msisdn",LongType(),True),
StructField("is_real",IntegerType(),True),
StructField("apps_id",IntegerType(),True),
StructField("mb_t1",DoubleType(),True),
StructField("mb_t2",DoubleType(),True),
StructField("mb_t3",DoubleType(),True),
StructField("mb_t4",DoubleType(),True),
StructField("mb_t5",DoubleType(),True),
StructField("mb_t6",DoubleType(),True),
StructField("mb_t7",DoubleType(),True),
StructField("mb_t8",DoubleType(),True),
StructField("mb",DoubleType(),True),
StructField("mb_dl",DoubleType(),True),
StructField("year",IntegerType(),True),
StructField("month",IntegerType(),True),
StructField("day",IntegerType(),True)
])
apps = pd.read_parquet('orion_sample.parquet',engine='fastparquet')
df_ada_orion_apps_data = spark.createDataFrame(apps)
Here's the result
---------------------------------------------------------------------------
TypeError Traceback (most recent call last)
<ipython-input-3-db4542633b58> in <module>
1 apps = pd.read_parquet('orion_sample.parquet',engine='fastparquet')
----> 2 df_ada_orion_apps_data = spark.createDataFrame(apps)
/opt/cloudera/parcels/CDH-7.1.3-1.cdh7.1.3.p0.4992530/lib/spark/python/pyspark/sql/session.py in createDataFrame(self, data, schema, samplingRatio, verifySchema)
746 rdd, schema = self._createFromRDD(data.map(prepare), schema, samplingRatio)
747 else:
--> 748 rdd, schema = self._createFromLocal(map(prepare, data), schema)
749 jrdd = self._jvm.SerDeUtil.toJavaArray(rdd._to_java_object_rdd())
750 jdf = self._jsparkSession.applySchemaToPythonRDD(jrdd.rdd(), schema.json())
/opt/cloudera/parcels/CDH-7.1.3-1.cdh7.1.3.p0.4992530/lib/spark/python/pyspark/sql/session.py in _createFromLocal(self, data, schema)
414
415 if schema is None or isinstance(schema, (list, tuple)):
--> 416 struct = self._inferSchemaFromList(data, names=schema)
417 converter = _create_converter(struct)
418 data = map(converter, data)
/opt/cloudera/parcels/CDH-7.1.3-1.cdh7.1.3.p0.4992530/lib/spark/python/pyspark/sql/session.py in _inferSchemaFromList(self, data, names)
346 warnings.warn("inferring schema from dict is deprecated,"
347 "please use pyspark.sql.Row instead")
--> 348 schema = reduce(_merge_type, (_infer_schema(row, names) for row in data))
349 if _has_nulltype(schema):
350 raise ValueError("Some of types cannot be determined after inferring")
/opt/cloudera/parcels/CDH-7.1.3-1.cdh7.1.3.p0.4992530/lib/spark/python/pyspark/sql/types.py in _merge_type(a, b, name)
1099 fields = [StructField(f.name, _merge_type(f.dataType, nfs.get(f.name, NullType()),
1100 name=new_name(f.name)))
-> 1101 for f in a.fields]
1102 names = set([f.name for f in fields])
1103 for n in nfs:
/opt/cloudera/parcels/CDH-7.1.3-1.cdh7.1.3.p0.4992530/lib/spark/python/pyspark/sql/types.py in <listcomp>(.0)
1099 fields = [StructField(f.name, _merge_type(f.dataType, nfs.get(f.name, NullType()),
1100 name=new_name(f.name)))
-> 1101 for f in a.fields]
1102 names = set([f.name for f in fields])
1103 for n in nfs:
/opt/cloudera/parcels/CDH-7.1.3-1.cdh7.1.3.p0.4992530/lib/spark/python/pyspark/sql/types.py in _merge_type(a, b, name)
1092 elif type(a) is not type(b):
1093 # TODO: type cast (such as int -> long)
-> 1094 raise TypeError(new_msg("Can not merge type %s and %s" % (type(a), type(b))))
1095
1096 # same type
TypeError: field apps_id: Can not merge type <class 'pyspark.sql.types.LongType'> and <class 'pyspark.sql.types.StructType'>