如何解决Pyspark - 在两个 String 列上合并导致错误 - 无法转换为 struct？

我有两个具有以下架构的数据框

api_log_user_df

   root
 |-- action: string (nullable = true)
 |-- battleID: string (nullable = true)
 |-- chosenGrade: string (nullable = true)
 |-- classCode: string (nullable = true)
 |-- curriculumTreeID: string (nullable = true)
 |-- details: struct (nullable = true)
 |    |-- extensions: struct (nullable = true)
 |    |    |-- code: string (nullable = true)
 |    |    |-- exception: struct (nullable = true)
 |    |    |    |-- type: string (nullable = true)
 |    |    |-- type: string (nullable = true)
 |    |-- locations: struct (nullable = true)
 |    |    |-- column: long (nullable = true)
 |    |    |-- line: long (nullable = true)
 |    |-- message: string (nullable = true)
 |    |-- path: string (nullable = true)
 |-- duration: long (nullable = true)
 |-- env: string (nullable = true)
 |-- err: struct (nullable = true)
 |    |-- code: string (nullable = true)
 |    |-- message: string (nullable = true)
 |    |-- name: string (nullable = true)
 |    |-- stack: string (nullable = true)
 |-- error: struct (nullable = true)
 |    |-- code: string (nullable = true)
 |    |-- message: string (nullable = true)
 |    |-- requestId: string (nullable = true)
 |    |-- retryDelay: float (nullable = true)
 |    |-- retryable: boolean (nullable = true)
 |    |-- statusCode: long (nullable = true)
 |    |-- time: timestamp (nullable = true)
 |-- giftBoxID: string (nullable = true)
 |-- grade: string (nullable = true)
 |-- hostname: string (nullable = true)
 |-- key: long (nullable = true)
 |-- level: long (nullable = true)
 |-- message: string (nullable = true)
 |-- msg: string (nullable = true)
 |-- name: string (nullable = true)
 |-- parameters: struct (nullable = true)
 |    |-- DAPROPS: string (nullable = true)
 |    |-- userAgent: string (nullable = true)
 |-- pid: long (nullable = true)
 |-- properties: struct (nullable = true)
 |    |-- ExpressionAttributeValues: struct (nullable = true)
 |    |    |-- :giftBoxValue: struct (nullable = true)
 |    |    |    |-- M: struct (nullable = true)
 |    |    |    |    |-- ID: struct (nullable = true)
 |    |    |    |    |    |-- N: string (nullable = true)
 |    |    |    |    |-- isOpen: struct (nullable = true)
 |    |    |    |    |    |-- BOOL: boolean (nullable = true)
 |    |-- Key: struct (nullable = true)
 |    |    |-- id: struct (nullable = true)
 |    |    |    |-- N: string (nullable = true)
 |    |-- ReturnValues: string (nullable = true)
 |    |-- TableName: string (nullable = true)
 |    |-- UpdateExpression: string (nullable = true)
 |-- time: timestamp (nullable = true)
 |-- userID: string (nullable = true)
 |-- v: long (nullable = true)
 |-- _Metadata: map (nullable = true)
 |    |-- key: string
 |    |-- value: string (valueContainsNull = true)
 |-- calendar_date: date (nullable = true)
 |-- _id: string (nullable = true)
 |-- user_id: string (nullable = true)
 |-- FailedRules: struct (nullable = true)
 |    |-- characterSize: struct (nullable = true)
 |    |    |-- message: string (nullable = true)
 |    |    |-- property: string (nullable = true)
 |    |    |-- value: string (nullable = true)
 |    |-- gold: struct (nullable = true)
 |    |    |-- message: string (nullable = true)
 |    |    |-- property: string (nullable = true)
 |    |    |-- value: string (nullable = true)
 |    |-- level: struct (nullable = true)
 |    |    |-- message: string (nullable = true)
 |    |    |-- property: string (nullable = true)
 |    |    |-- value: string (nullable = true)
 |    |-- orbTiers: struct (nullable = true)
 |    |    |-- message: string (nullable = true)
 |    |    |-- property: string (nullable = true)
 |    |    |-- value: struct (nullable = true)
 |    |    |    |-- orb: long (nullable = true)
 |    |    |    |-- orbTier: long (nullable = true)
 |    |    |    |-- tier: long (nullable = true)
 |    |-- stars: struct (nullable = true)
 |    |    |-- message: string (nullable = true)
 |    |    |-- property: string (nullable = true)
 |    |    |-- value: string (nullable = true)

api_log_alt_user_df

_id:string
alt_userID:string

我正在尝试像下面那样加入这两个 DF

api_log_df = (
    api_log_user_df.join(api_log_alt_user_df,on="_id",how="left")
    .drop("_Metadata")
    .withColumn(
        "__user_id",F.coalesce(
            F.coalesce(F.col("userID"),F.col("user_id").cast("string")),F.col("alt_userID"),),)
)

我收到如下错误消息，但我不明白为什么？这里的所有列都是 (coalesce) 数据类型字符串。我不确定如何处理此错误以进行调试。

org.apache.spark.SparkException: Job aborted due to stage failure: Task 13 in stage 11162.0 Failed 4 times,most recent failure: Lost task 13.3 in stage 11162.0 (TID 10779,10.88.249.71,executor 1): java.lang.IllegalArgumentException: The value ([]) of the type (java.util.ArrayList) cannot be converted to struct<code:string,exception:struct<type:string>,type:string>
    at org.apache.spark.sql.catalyst.CatalystTypeConverters$StructConverter.toCatalystImpl(CatalystTypeConverters.scala:262)

仅在某些日子每天运行时不会出现此错误，但我无法找出可能导致该错误的数据类型。

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）

val spark = SparkSession.builder()
      .master("local")
      .appName("app")
      .config("spark.mongodb.input.uri",uri)
      .config("spark.mongodb.input.collection","collectionName")
      .config("spark.mongodb.input.readPreference.name","secondary")
      .getorCreate()

val df = MongoSpark.load(spark).limit(1)

root
 |-- A: struct (nullable = true)
 |    |-- oid: string (nullable = true)
 |-- B: boolean (nullable = true)
 |-- C: string (nullable = true)
 |-- D: string (nullable = true)
 |-- E: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- a: string (nullable = true)
 |    |    |-- b: string (nullable = true)
 |    |    |-- c: string (nullable = true)
 |    |    |-- d: string (nullable = true)

Cannot cast STRING into a StructType(StructField(a,StringType,true),StructField(b,StructField(c,StructField(d,true)) (value: BsonString{value=''http://...some url...''})

{"Timestamp":"2015-01-01T00:00:06.222Z"}

SparkSession
  .builder()
  .master("local[*]")
  .appName("my-app")
  .getorCreate()
  .readStream()
  .format("kafka")
  ... //some magic
  .writeStream()
  .format("console")
  .start()
  .awaitTermination();

StructType schema = DataTypes.createStructType(new StructField[] {
        DataTypes.createStructField("Timestamp",DataTypes.TimestampType,true),});

+--------------------+
|           Timestamp|
+--------------------+
|2015-01-01 01:00:...|
|2015-01-01 01:00:...|
+--------------------+

spark.conf.set("spark.sql.session.timeZone","UTC")

select *,cast('2017-01-01 10:10:10' as timestamp) from soMetable