您可以通过对列进行分组并将其收集为对象列表来使用以下转换。首先，您需要读取CSV文件并执行连接操作。我们将从生成的连接数据开始。

from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DateType, TimestampType
schema = StructType([
    StructField("TableName", StringType(), True),
    StructField("DC", StringType(), True),
    StructField("Owner", StringType(), True),
    StructField("LT", StringType(), True),
    StructField("Sour", StringType(), True),
    StructField("SourceType", StringType(), True),
    StructField("SourceConn", StringType(), True),
    StructField("ColumnName", StringType(), True),
    StructField("ColPos", IntegerType(), True),
    StructField("DataType", StringType(), True),
    StructField("IsPrimary", IntegerType(), True),
])
data = [
    ("A", "Confidential", "HR", "Full", "S1", "SQL DB", "<XYZ>", "A1", 1, "string",1),
    ("A", "Confidential", "HR", "Full", "S1", "SQL DB", "<XYZ>", "A2", 2, "int",0),
    ("A", "Confidential", "HR", "Full", "S1", "SQL DB", "<XYZ>", "A3", 3, "date",1),
    ("A", "Confidential", "HR", "Full", "S1", "SQL DB", "<XYZ>", "A4", 4, "timestamp",0),
]
df = spark.createDataFrame(data, schema=schema)
df.display()

字符串
要获得预期的结果，请使用以下转换查询。

from pyspark.sql.functions import collect_list, struct, col
result_df = (
    df.groupBy("TableName", "LT", "DC", "Owner", "Sour", "SourceType", "SourceConn")
    .agg(
        collect_list(
            struct("ColumnName", "ColPos", "DataType", "IsPrimary")
        ).alias("SourceObjectDetails")
    )
    .select(
        "TableName",
        "LT",
        "DC",
        "Owner",
        struct(
            col("Sour").alias("Name"),
            col("SourceType").alias("Type"),
            col("SourceConn").alias("ConnectDetails"),
            "SourceObjectDetails"
        ).alias("Source")
    )
)
result_df.display()

型
要为列表中的每个PublicationObject获取JSON对象，请使用以下代码。

import json
jsons = [json.loads(i) for i in result_df.toJSON().collect()]

型
输出量：

[{'TableName': 'A',
  'LT': 'Full',
  'DC': 'Confidential',
  'Owner': 'HR',
  'Source': {'Name': 'S1',
   'Type': 'SQL DB',
   'ConnectDetails': '<XYZ>',
   'SourceObjectDetails': [{'ColumnName': 'A1',
     'ColPos': 1,
     'DataType': 'string',
     'IsPrimary': 1},
    {'ColumnName': 'A2', 'ColPos': 2, 'DataType': 'int', 'IsPrimary': 0},
    {'ColumnName': 'A3', 'ColPos': 3, 'DataType': 'date', 'IsPrimary': 1},
    {'ColumnName': 'A4',
     'ColPos': 4,
     'DataType': 'timestamp',
     'IsPrimary': 0}]}}]

型