I am trying to extract first element of probability column (vector data type) using UDF in Pyspark. I was able to get the new dataframe with extracted values in probability column. I also checked the data type of probability column which has changed from vector to float. But I am not able to write the dataframe into hive table. I get numpy module not found error. Deploy mode is client. Is there a workaround other than installing numpy in all the worker nodes ?
Code -
spark = (SparkSession
.builder
.appName("Model_Scoring")
.master('yarn')
.enableHiveSupport()
.getOrCreate()
)
hive = HiveWarehouseSession.session(spark).build()
hive.setDatabase("hc360_models")
final_ads = spark.read.parquet("hdfs://DATAHUB/datahube/feature_engineering/final_ads.parquet")
model = PipelineModel.load("/tmp/fitted_model_new/")
first_element=udf(lambda v:float(v[0]),FloatType())
out = model.transform(final_ads)
out = out.withColumn("probability",first_element("probability")).drop('features').drop('rawPrediction')
out.show(10)
out.write.mode("append").format(HiveWarehouseSession.HIVE_WAREHOUSE_CONNECTOR).option("table","test_hypertension_table_final3").save()
spark.stop()
Error -
ImportError: ('No module named numpy', <function _parse_datatype_json_string at 0x7f83370922a8>, (u'{"type":"struct","fields":[{"name":"_0","type":{"type":"udt","class":"org.apache.spark.ml.linalg.VectorUDT","pyClass":"pyspark.ml.linalg.VectorUDT","sqlType":{"type":"struct","fields":[{"name":"type","type":"byte","nullable":false,"metadata":{}},{"name":"size","type":"integer","nullable":true,"metadata":{}},{"name":"indices","type":{"type":"array","elementType":"integer","containsNull":false},"nullable":true,"metadata":{}},{"name":"values","type":{"type":"array","elementType":"double","containsNull":false},"nullable":true,"metadata":{}}]}},"nullable":true,"metadata":{}},{"name":"_1","type":"integer","nullable":true,"metadata":{}}]}',))
Sample data -
Schema -
StructType(List(StructField(patient_id,IntegerType,true), StructField(carrier_operational_id,IntegerType,false), StructField(gender_cde,StringType,true), StructField(pre_fixed_mpr_qty,DecimalType(38,8),true), StructField(idx_days_in_gap,DecimalType(11,1),true), StructField(age,DecimalType(6,1),true), StructField(post_fixed_mpr_adh_ind,DecimalType(2,1),true), StructField(probability,FloatType,true), StructField(prediction,DoubleType,false), StructField(run_date,TimestampType,false)))