PySpark UDF，输入时仅无值

Question

我的Kafka流媒体应用程序中的UDF函数有问题。每次调用UDF函数时，输入上只有None值，而不是有效的列值。打字错误 然后引发，因为应用程序需要str，而不是None

自定义函数定义：

@udf(returnType=StringType())
def get_asn(ip_addr):
    from fm_kafka2parquet.asn_lookup import AsnLookup

    result = AsnLookup\
        .get_instance(ASN_DB_PATH)\
        .get().lookup(ip_addr)[0]  # first record from tuple is ASN number
    if result is None:
        return "n/a"
    return result

UDF函数调用：

  # data frame for netflow reading
  df = spark \
      .readStream \
      .format("kafka") \
      .option("kafka.bootstrap.servers", CONFIG_KAFKA_BOOTSTRAP) \
      .option("subscribe", CONFIG_KAFKA_TOPIC) \
      .option("startingOffsets", "latest") \
      .load() \
      .selectExpr("CAST(value AS STRING)") \
      .withColumn("net", from_json("value", Structures.get_ipfix_structure())) \
      .select("net.*")

  # remove ipfix prefix in case of ipfixv1 collector
  temp_list = []
  for c in df.columns:
      new_name = c.replace('ipfix.', '')
      temp_list.append(new_name)
  df = df.toDF(*temp_list)

  # enrichment
  edf = df \
      .withColumn("sourceAS", get_asn('sourceIPv4Address')) \
      .withColumn("destinationAS", get_asn('destinationIPv4Address'))

一切都以err结束，err由get_asn UDF函数使用的pyasn库引发：

TypeError: search_best() argument 1 must be str, not None

Answer 1

试着像下面提到的那样使用它。 .withColumn（“sourceAS”，get_asn（F.col（'sourceIPv4Address'））

Answer 2

而且，这看起来很可疑

# remove ipfix prefix in case of ipfixv1 collector
  temp_list = []
  for c in df.columns:
      new_name = c.replace('ipfix.', '')
      temp_list.append(new_name)
  df = df.toDF(*temp_list)

您正在更改列名，然后选择它们，但新列名不在数据框中，对吗？因此，它必须返回空数据帧

如果要重命名列，请使用-

df = df.withColumnRenamed(c, c.replace('ipfix.', ''))

有关如何在pyspark中清除列名的详细信息，请参阅此-https://www.youtube.com/watch?v=vAHPAP9Oagc&t=1s