Итак, у меня есть пользовательский df с уникальными user_ids и второй df с набором вопросов. Затем я хотел бы объединить dfs, чтобы каждый user_id был прикреплен к полному набору вопросов:
Пользователь Df:
+--------------------------+
|user_id |
+--------------------------+
|GDDVWWIOOKDY4WWBCICM4VOQHQ|
|77VC23NYEWLGHVVS4UMHJEVESU|
|VCOX7HUHTMPFCUOGYWGL4DMIRI|
|XPJBJMABYXLTZCKSONJVBCOXQM|
|QHTPQSFNOA5YEWH6N7FREBMMDM|
|JLQNBYCSC4DGCOHNLRBK5UANWI|
|RWYUOLBKIQMZVYHZJYCQ7SGTKA|
|CR33NGPK2GKK6G35SLZB7TGIJE|
|N6K7URSGH65T5UT6PZHMN62E2U|
|SZMPG3FQQOHGDV23UVXODTQETE|
+--------------------------+
Вопросы Df
+--------------------+-------------------+-----------------+--------------------+
| category_type| category_subject| question_id| question|
+--------------------+-------------------+-----------------+--------------------+
|Consumer & Lifestyle| Dietary Habits|pdl_diet_identity|Eating habits des...|
|Consumer & Lifestyle| Dietary Habits|pdl_diet_identity|Eating habits des...|
|Consumer & Lifestyle| Dietary Habits|pdl_diet_identity|Eating habits des...|
|Consumer & Lifestyle| Dietary Habits|pdl_diet_identity|Eating habits des...|
|Consumer & Lifestyle| Dietary Habits|pdl_diet_identity|Eating habits des...|
|Consumer & Lifestyle| Dietary Habits|pdl_diet_identity|Eating habits des...|
|Consumer & Lifestyle| Dietary Habits|pdl_diet_identity|Eating habits des...|
| Demographics|Social Demographics|pdl_ethnicity_new| Ethnicity|
| Demographics|Social Demographics|pdl_ethnicity_new| Ethnicity|
| Demographics|Social Demographics|pdl_ethnicity_new| Ethnicity|
+--------------------+-------------------+-----------------+--------------------+
Так что в данный момент я превращаю user_ids в список и перебираю их, создавая новую колонку по вопросам df, создавая временную df из результатов. Затем я объединяюсь в окончательный df, чтобы сохранить результаты для этой итерации user_id, как показано ниже:
создать список user_id:
unique_users_list = users_df \
.select("user_id") \
.agg(f.collect_list('user_id')).collect()[0][0]
создать пустой финальный df для добавления к:
finaldf_schema = StructType([
StructField("category_type", StringType(), False),
StructField("category_subject", StringType(), False),
StructField("question_id", StringType(), False),
StructField("question", StringType(), False),
StructField("user_id", StringType(), False)
])
final_df = spark.createDataFrame([], finaldf_schema)
Затем выполните цикл по user_id, объединяющемуся с вопросами df:
for user_id in unique_users_list:
temp_df = questions_df.withColumn("user_id", f.lit(user_id))
final_df = final_df.union(temp_df)
Однако я считаю, что производительность очень низкая. Есть ли более эффективный и быстрый способ сделать это, пожалуйста.
Спасибо