Создание нового rdd из другого rdd в Python

Question

Я совершенно новичок в pysparks и rdd. Я пытаюсь понять, как работает rdd, и у меня возникают проблемы с доступом к части данных в rdd. Я хотел бы выбрать несколько столбцов из существующего rdd и создать новый rdd.

Пример ниже:

user_rdd = [Row(id=u'1', first_name=u'Steve', last_name=u'Kent', email=u's.kent@email.com'),Row(id=u'2', first_name=u'Margaret', last_name=u'Peace', email=u'marge.peace@email.com')] 

display(user_rdd) 

| email     | first_name | id | last_name 
| s.kent@email.com  | Steve  | 1 | Kent 
| marge.peace@email.com | Margaret | 2 | Peace 

Как выбрать 2 колонки из user_rdd и создать новый RDD, как показано ниже?

| id | first_name | last_name | full_name 
| 1 | Steve  | Kent  | Steve Kent 
| 2 | Margaret | Peace  | Margaret Peace 

Answer 1

Вот, наверное, самый простой способ сделать то, что вы после (хотя ваш РДД выглядит, как он был получен из DataFrame)

from pyspark.sql import Row 

user_rdd = sc.parallelize([ 
    Row(id=u'1', 
     first_name=u'Steve', 
     last_name=u'Kent', 
     email=u's.kent@email.com'), 
    Row(id=u'2', 
     first_name=u'Margaret', 
     last_name=u'Peace', 
     email=u'marge.peace@email.com') 
]) 

new_rdd = user_rdd.map(lambda row: Row(
     first_name=row.first_name, 
     last_name=row.last_name, 
     full_name=row.first_name + ' ' + row.last_name)) 

new_rdd.take(2) 

Answer 2

Вы можете попробовать это:

sc = SparkContext(conf=conf) 
user_rdd = [Row(id=u'1', first_name=u'Steve', last_name=u'Kent', email=u's.kent@email.com'), 
      Row(id=u'2', first_name=u'Margaret', last_name=u'Peace', email=u'marge.peace@email.com')] 
rdd = sc.parallelize(user_rdd).map(lambda x: Row(fullname=' '.join([x['first_name'], x['last_name']]), **x.asDict())) 
print(rdd.collect()) 

Answer 3

Не знаю, почему вы хотите сделать это через РД, так как список строк, по существу DataFrame и легче выбрать столбец и сцепить его там. Это также более эффективно, чем использование RDD.

from pyspark.sql import Row 
from pyspark.sql.functions import concat_ws 

user_rdd = [Row(id=u'1', first_name=u'Steve', last_name=u'Kent', email=u's.kent@email.com'),Row(id=u'2', first_name=u'Margaret', last_name=u'Peace', email=u'marge.peace@email.com')] 

user_df = spark_session.createDataFrame(user_rdd) 

user_df.select(user_df.id, user_df.first_name, user_df.last_name, concat_ws(' ',user_df.first_name, user_df.last_name).alias('full_name')).show() 

Это даст вам результат, как DataFrame:

+---+----------+---------+--------------+ 
| id|first_name|last_name|  full_name| 
+---+----------+---------+--------------+ 
| 1|  Steve|  Kent| Steve Kent| 
| 2| Margaret| Peace|Margaret Peace| 
+---+----------+---------+--------------+