如何在PySpark中按sourcemmsi分组导出指定格式文本文件

完全可以用PySpark实现这个需求，以下是具体步骤和代码示例：

1. 分组收集坐标数据 #

首先按sourcemmsi字段分组，将每个分组下的所有coordinates字符串收集为一个列表：

from pyspark.sql import functions as F

# 假设你的DataFrame名为df
grouped_df = df.groupBy("sourcemmsi").agg(
    F.collect_list("coordinates").alias("coords_list")
)

2. 格式化目标文本行 #

通过字符串拼接，将分组后的数据格式化为你需要的{sourcemmsi, 坐标1, 坐标2...}格式：

formatted_df = grouped_df.withColumn(
    "line",
    F.concat(
        F.lit("{"),
        F.col("sourcemmsi").cast("string"),
        F.lit(", "),
        F.concat_ws(", ", "coords_list"),
        F.lit(" }")
    )
)

3. 保存为文本文件 #

选择格式化后的line列，写入到指定目录：

# 替换为你的输出路径
output_path = "/your/output/directory"
formatted_df.select("line").write.mode("overwrite").text(output_path)

补充：如果需要给坐标添加第三个数值 #

如果你的示例中坐标的第三个值（如43.43）是需要额外补充的（原数据中没有），可以通过自定义UDF处理坐标字符串：

from pyspark.sql.types import StringType

def append_third_value(coord_str):
    # 原坐标格式为"(x,y)"，修改为"(x,y, value)"，这里value可替换为实际需要的数值
    return coord_str[:-1] + ", 0.0)"

# 注册UDF
append_third_udf = F.udf(append_third_value, StringType())

# 先处理坐标列，添加第三个值
df_processed = df.withColumn("coordinates_3d", append_third_udf("coordinates"))

# 后续分组、格式化步骤同前
grouped_df = df_processed.groupBy("sourcemmsi").agg(F.collect_list("coordinates_3d").alias("coords_list"))
formatted_df = grouped_df.withColumn(
    "line",
    F.concat(F.lit("{"), F.col("sourcemmsi").cast("string"), F.lit(", "), F.concat_ws(", ", "coords_list"), F.lit(" }"))
)

# 保存
formatted_df.select("line").write.mode("overwrite").text(output_path)

注意事项 #

• 如果需要生成单个文本文件，可以在保存前添加coalesce(1)（如formatted_df.select("line").coalesce(1).write...），但大数据量下不建议使用，会影响性能。
• write.mode("overwrite")会覆盖目标目录的已有数据，可根据需求调整为append或ignore。

内容的提问来源于stack exchange，提问作者user19384320