如果您正在使用Spark 2.x,我相信这就是您需要的:
from pyspark.sql.functions import udf
from pyspark.mllib.linalg import Vectors
from pyspark.ml.linalg import VectorUDT
from pyspark.ml.feature import StringIndexer
df = spark.createDataFrame([("[-0.38475, 0.568]", "label1"), ("[0.645734, 0.699]", "label2")], ("features", "label"))
def parse(s):
try:
return Vectors.parse(s).asML()
except:
return None
parse_ = udf(parse, VectorUDT())
parsed = df.withColumn("features", parse_("features"))
indexer = StringIndexer(inputCol="label", outputCol="label_indexed")
indexer.fit(parsed).transform(parsed).show()
在Spark 1.6版本中,与之前并没有太大区别:
from pyspark.sql.functions import udf
from pyspark.ml.feature import StringIndexer
from pyspark.mllib.linalg import Vectors, VectorUDT
df = sqlContext.createDataFrame([("[-0.38475, 0.568]", "label1"), ("[0.645734, 0.699]", "label2")], ("features", "label"))
parse_ = udf(Vectors.parse, VectorUDT())
parsed = df.withColumn("features", parse_("features"))
indexer = StringIndexer(inputCol="label", outputCol="label_indexed")
indexer.fit(parsed).transform(parsed).show()
Vectors
拥有一个parse
函数,可以帮助您实现您想要做的事情。