您可以创建一个列数组，并将其传递给withColumn中的udf。

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
import json
# Sample data
data = [
    ("2022-01-01", "192.168.1.1", "00:1A:2B:3C:4D:5E"),
    ("2022-01-02", "192.168.1.2", "11:22:33:44:55:66"),
    # Add more rows with similar structure
]
# Define the schema for the DataFrame
columns = ["when", "ip", "mac"]
# Create the DataFrame
df = spark.createDataFrame(data, columns)
# Define a UDF to create a dictionary from variable parameters
@udf(StringType())
def create_json_packet(*args):
    subset_dict = dict(zip(subset_columns, args))
    return json.dumps(subset_dict)
# Specify the subset of columns you want to include in the dictionary
subset_columns = ["when", "ip"]
df \
.withColumn("output_json", create_json_packet(*subset_columns)) \
.show(10, truncate=False)
Output:
+----------+-----------+-----------------+-------------------------------------------+
|when      |ip         |mac              |output_json                                |
+----------+-----------+-----------------+-------------------------------------------+
|2022-01-01|192.168.1.1|00:1A:2B:3C:4D:5E|{"when": "2022-01-01", "ip": "192.168.1.1"}|
|2022-01-02|192.168.1.2|11:22:33:44:55:66|{"when": "2022-01-02", "ip": "192.168.1.2"}|
+----------+-----------+-----------------+-------------------------------------------+

字符串
或者，如果你愿意，你也可以把subset_columns数组作为参数传递给udf：

from pyspark.sql.functions import lit
# Define a UDF to create a dictionary from variable parameters
@udf(StringType())
def create_json_packet(subset_columns, *args):
    subset_dict = dict(zip(subset_columns, args))
    return json.dumps(subset_dict)
df \
.withColumn("output_json", create_json_packet(lit(subset_columns), *subset_columns)) \
.show(10, truncate=False)

型