pySpark如何访问（键，元组）RDD（python）中元组中的值答案

【问题标题】：pySpark how to access the values in a tuple in a (key,tuple) RDD (python)pySpark如何访问（键，元组）RDD（python）中元组中的值
【发布时间】：2017-04-04 14:15:34
【问题描述】：

我正在尝试访问 PipelineRDD 中包含的值这是我开始的：

1. RDD = (key,code,value)

data = [(11720, (u'I50800', 0.08229813664596274)), (11720, (u'I50801', 0.03076923076923077))]

*强调文字*2。我需要它按第一个值分组并将其转换为 (key,tuple ) where tuple = (code,value)

testFeatures = lab_FeatureTuples = labEvents.select('ITEMID', 'SUBJECT_ID','NORM_ITEM_CNT')\ .orderBy('SUBJECT_ID','ITEMID')\ .rdd.map(lambda (ITEMID,SUBJECT_ID,NORM_ITEM_CNT):(SUBJECT_ID,(ITEMID,NORM_ITEM_CNT)))\ .groupByKey()

testFeatures =  [(11720, [(u'I50800', 0.08229813664596274)),  (u'I50801', 0.03076923076923077)])]

在元组 = (code,value) 上，我想得到以下内容：

从中创建一个 sparseVector，以便我可以将其用于 SVM 模型

结果.take(1)

【问题讨论】：

请正确重新格式化您的代码

标签： python vector pyspark svm rdd

【解决方案1】：

这是一种方法：

import pyspark
import pyspark.sql.functions as sf
import pyspark.sql.types as sparktypes
sc = pyspark.SparkContext()
sqlc = pyspark.SQLContext(sc)

data = [(11720, (u'I50800', 0.08229813664596274)), 
        (11720, (u'I50801', 0.03076923076923077))]
rdd = sc.parallelize(data)

df = sqlc.createDataFrame(rdd,  ['idx', 'tuple'])
df.show()

给予，

+-----+--------------------+
|  idx|               tuple|
+-----+--------------------+
|11720|[I50800,0.0822981...|
|11720|[I50801,0.0307692...|
+-----+--------------------+

现在定义 pyspark 用户定义函数：

extract_tuple_0 = sf.udf(lambda x: x[0], returnType=sparktypes.StringType())
extract_tuple_1 = sf.udf(lambda x: x[1], returnType=sparktypes.FloatType())
df = df.withColumn('tup0', extract_tuple_0(sf.col('tuple')))

df = df.withColumn('tup1', extract_tuple_1(sf.col('tuple')))
df.show()

给予：

+-----+--------------------+----------+------+
|  idx|               tuple|      tup1|  tup0|
+-----+--------------------+----------+------+
|11720|[I50800,0.0822981...|0.08229814|I50800|
|11720|[I50801,0.0307692...|0.03076923|I50801|
+-----+--------------------+----------+------+

【讨论】：