如何复制源 Spark 数据帧架构的可空性状态并将其强制到目标 Spark 数据帧上？

Question

我正在使用数据库。假设我有两个Spark Dataframes（我正在使用PySpark）：

< li>df_source < li>df_target

如果df_source具有以下模式:

root
 |-- name: string (nullable = false)
 |-- id: long (nullable = false)
 |-- age: long (nullable = true)

df_target具有以下架构：

root
 |-- name: string (nullable = true)
 |-- id: long (nullable = false)
 |-- age: long (nullable = false)

如何有效地创建另一个数据帧，df_final其中可以将df_source中的（null = true/false）属性强制到df_target？

我尝试了以下方法:

df_final = spark.createDataFrame(df_target.rdd, schema = df_source.schema)

通过这种方法，我能够获得所需的结果，但是对于我拥有的数据集大小来说，它似乎需要很长时间。对于较小的数据集，它可以正常工作。对于较大的数据集，使用 collect（） 函数而不是 rdd 转换显然更糟。

我想指出的是，我在这里要做的唯一一件事是从源模式中复制可空性部分，并在目标中对其进行相应的更改，以获得最终的数据帧。

是否有一种方法可以执行某种类型的可空性转换，在性能方面类似于.withColumn（），不需要RDD转换，也不需要代码中明确的列名规范？列顺序已在源和目标之间对齐。

附加上下文：我需要这样做的原因是因为我需要使用Spark BQ连接器将df_final写入（追加）到Google BigQuery表。因此，即使我的Spark Dataframe在列中没有任何空值，但null属性设置为true，BigQuery表也会拒绝写入操作，因为BigQuery表中的该列可能将null属性设置为false，并且模式不匹配。

Answer 1

由于您知道age不能为空，因此可以将合并age和常量文本以创建不可为空的字段。对于＜code＞可为空的＜code＞字段必须从＜code＞false＜/code＞转换为＜code＞true＜/code＞的字段，可以使用＜code＞表达式。

from pyspark.sql.types import StructType, StructField, StringType, LongType
from pyspark.sql import functions as F

df_source_schema = StructType([
    StructField("name", StringType(), False),
    StructField("id", LongType(), False),
    StructField("age", LongType(), True),
])

df_target_schema = StructType([
    StructField("name", StringType(), True),
    StructField("id", LongType(), False),
    StructField("age", LongType(), False),
])

df_source = spark.createDataFrame([("a", 1, 18, ), ], df_source_schema)
df_source.printSchema()
"""
root
 |-- name: string (nullable = false)
 |-- id: long (nullable = false)
 |-- age: long (nullable = true)
"""

df_target = spark.createDataFrame([("a", 1, 18), ], df_target_schema)
df_target.printSchema()
"""
root
 |-- name: string (nullable = true)
 |-- id: long (nullable = false)
 |-- age: long (nullable = false)
"""

# Construct selection expression based on the logic described above

target_field_nullable_map = {field.name: field.nullable for field in df_target.schema}

selection_expr = []

for src_field in df_source.schema:
    field_name = src_field.name
    field_type = src_field.dataType
    
    if target_field_nullable_map[field_name] != src_field.nullable:
        if src_field.nullable:
            selection_expr.append(F.when(F.col(field_name).isNotNull(), F.col(field_name)).otherwise(F.lit(None)).alias(field_name))
        else:
            selection_expr.append(F.coalesce(F.col(field_name), F.lit("-1").cast(field_type)).alias(field_name))
    else:
        selection_expr.append(F.col(field_name))

df_final = df_target.select(*selection_expr)
df_final.printSchema()
"""
root
 |-- name: string (nullable = false)
 |-- id: long (nullable = false)
 |-- age: long (nullable = true)
"""

对于要使合并表达式为 null，其子表达式的所有子表达式都必须为 null，从此处可以看出。因为 lit 是非空表达式，当值 ！= 空合并导致不可为空的列时。

当表达式可以为 null 时，如果 else 表达式可为 null，则任何分支都可以为 null，如此处所述。