How can you filter invalid IP addresses to the PySpark?

Decurro 0
Viewed 44

I have a table of IP addresses.

enter image description here

Through the Dropna () function I deleted Null elements.

I also removed repetitions and frequent addresses.

dt4new1 = dt4new.where("not (src = '0.0.0.0') and  not (dst = '0.0.0.0') ").where("not (src = '127.0.0.1') and  not (dst = '127.0.0.1') ").where("not (src = dst) ").where("not (src = '1')")

Now I have a table with pretty random names. How can I get rid of them?

enter image description here

pyspark apache-spark-sql
2 Answers

After all the removals for nulls and repetitions, you can flag the addresses that are valid, i.e. they have four set of numbers delimited by ..

Following is an example of flagging. The flag can be used to filter.

data_sdf. \
    withColumn('ip_split', func.split('ip', '\.')). \
    withColumn('valid_flag', 
               func.size(func.expr('filter(ip_split, x -> x between 0 and 255)')) == 4
               ). \
    show()

# +-----------+----------------+----------+
# |         ip|        ip_split|valid_flag|
# +-----------+----------------+----------+
# |       1.3"|         [1, 3"]|     false|
# |.194.32.167|[, 194, 32, 167]|     false|
# | 0.77.160.4| [0, 77, 160, 4]|      true|
# +-----------+----------------+----------+

It splits the string by . and then only retains the elements that fall between 0 - 999. Taking a count of the retained elements can give you the flag - if there are 4 retained elements, the ip address is valid.

samkart0

here's how i would write it. not sure what random names you are trying to get rid of based on the screenshot.

EDIT: you can try using a regex function in a UDF to validate the ips

from pyspark.sql import functions as F
from pyspark.sql.types import *
import re

# Make a regular expression
# for validating an Ip-address
regex = "^((25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])\.){3}(25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])$"

def check_ip(ip): 
# pass the regular expression
# and the string in search() method
    if(re.search(regex, ip)):
        print("Valid Ip address")
        return True         
    else:
        print("Invalid Ip address")
        return False

check_ip_udf = F.udf(check_ip, BooleanType())

dt4new1 = dt4new.filter(
    ~F.col("src").isin("0.0.0.0", "127.0.0.1", "1")
    & ~F.col("dst").isin("0.0.0.0", "127.0.0.1")
    & (F.col("src") != F.col("dst"))
    & (check_ip_udf(F.col("src")) == True)
    & (check_ip_udf(F.col("dst")) == True)
)

dt4new1.show()
iambdot0
Related