Pyspark:从Struct中识别arrayType列，并调用udf将数组转换为字符串

Question

我正在创建一个加速器，将数据从源迁移到目标。例如，我将从API中选择数据，并将数据迁移到csv。在将数据转换为csv时，我遇到了处理arraytype的问题。我使用了withColumn和concat_ws方法（即df1=df.WithColum（'薄膜'，F.concat_ ws（'：'，F.col（'薄膜“））薄膜是arraytype列）进行转换，并且它起了作用。现在我希望这是动态发生的。我的意思是，在不指定列名的情况下，是否有一种方法可以从具有arraytype的结构中选择列名，然后调用udf？

感谢您的宝贵时间！

Answer 1

您可以使用df.schema获取列的类型。根据列的类型，您可以应用concat_ws，也可以不应用:

data = [["test1", "test2", [1,2,3], ["a","b","c"]]]
schema= ["col1", "col2", "arr1", "arr2"]
df = spark.createDataFrame(data, schema)

array_cols = [F.concat_ws(":", c.name).alias(c.name) \
    for c in df.schema if isinstance(c.dataType, T.ArrayType) ]
other_cols = [F.col(c.name) \
    for c in df.schema if not isinstance(c.dataType, T.ArrayType) ]

df = df.select(other_cols + array_cols)

结果：

+-----+-----+-----+-----+
| col1| col2| arr1| arr2|
+-----+-----+-----+-----+
|test1|test2|1:2:3|a:b:c|
+-----+-----+-----+-----+