從這個問題開始到現在已經 3 周了。長話短說，我的目標是創建一個用于推薦的用戶專案矩陣（余弦，悲傷，......）。
為此，我創建了下面的代碼。第一個函式列出了我的 HDFS 中的所有檔案，這樣我就可以同時讀取我的所有訂單、我的產品視圖和我添加到購物車的產品。
這些操作作業正常，我可以毫無問題地列印我的資料框。當我開始對我的 3 個 DF 的并集執行 .distinct 之類的操作時，我無法在我的 df 上做任何事情，我可以展示、收集、toPandas，而不會出現我無法理解的大錯誤。
閱讀鑲木地板時出錯，但在閱讀之后我可以正確閱讀我的 df，所以閱讀沒有問題嗎？
我已經嘗試過，寫傳統格式，還有很多其他的東西。
你們中的一些人可以幫助我嗎？（隨時提出任何問題，或索取更多代碼/檔案）
謝謝 ！阿德里安
def get_data_raw_to_df(spark, nb_months=36, id_shop="-1"):
'''get Data'''
filename_list_paid = get_date_list(
spark,
"hdfs:///" id_shop "/master/order",
id_shop,
nb_months,
"paid_"
)
filename_list_cart = get_date_list(
spark,
"hdfs:///" id_shop "/master/product",
id_shop,
nb_months,
"cart_"
)
filename_list_view = get_date_list(
spark,
"hdfs:///" id_shop "/master/product",
id_shop,
nb_months,
"view_"
)
empty_rdd = spark.sparkContext.emptyRDD()
schema = StructType([
StructField('cid', StringType(), True),
StructField('sid', StringType(), True),
StructField('product_id', StringType(), True),
StructField('buy_product', IntegerType(), True),
StructField('cart_product', IntegerType(), True),
StructField('view_product', IntegerType(), True),
])
df_users_items = empty_rdd.toDF(schema)
if len(filename_list_paid) > 0:
df_tmp_buy_product = spark.read.parquet(*filename_list_paid)\
.filter("is_bot == 'false'")\
.select('cid',
'sid',
F.explode('items'))\
.select('cid', 'sid',
col("col.product_id"))\
.withColumn("buy_product",
F.lit(5))
if len(filename_list_cart) > 0:
df_tmp_cart_product = spark.read.parquet(*filename_list_cart)\
.filter("is_bot == 'false'")\
.select('cid', 'sid',
'product_id')\
.withColumn("cart_product",
F.lit(2))
if len(filename_list_view) > 0:
df_tmp_view_product = spark.read.parquet(*filename_list_view)\
.filter("is_bot == 'false'")\
.select('cid', 'sid',
'product_id')\
.withColumn("view_product",
F.lit(1))
try:
# Setup product buy
df_users_items = df_users_items\
.unionByName(df_tmp_buy_product
.select('cid',
'product_id',
'sid',
'buy_product'),
allowMissingColumns=True)
# Can do thing form here
df_tmp = (df_tmp_buy_product.select('cid', 'sid')
.unionByName(
df_tmp_cart_product.select('cid', 'sid'))
.unionByName(
df_tmp_view_product.select('cid', 'sid'))).distinct()
# Errors from here
...
錯誤日志：https ://justpaste.it/2y7jc
uj5u.com熱心網友回復：
spark.read.option('mergeSchema', True).parquet(*filename_list_paid)可能有幫助。或者您可以嘗試閱讀此特定檔案。
用于spark.read.parquet('.../order/paid__20220324.parquet/part-00039-ff535f97*').printSchema()查看是否有String或其他型別。
轉載請註明出處，本文鏈接：https://www.uj5u.com/net/473774.html
標籤：Python 爪哇 阿帕奇火花 Hadoop pyspark