You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by "ashok34668@yahoo.com.INVALID" <as...@yahoo.com.INVALID> on 2023/09/15 18:13:00 UTC
Filter out 20% of rows
Hi team,
I am using PySpark 3.4
I have a table of million rows that has few columns. among them incoming ips and what is known as gbps (Gigabytes per second) and date and time of incoming ip.
I want to filter out 20% of low active ips and work on the remainder of data. How can I do thiis in PySpark?
Thanks
Re: Filter out 20% of rows
Posted by Bjørn Jørgensen <bj...@gmail.com>.
EDIT:
I don't think that the question asker will have only returned the top 25
percentages.
lør. 16. sep. 2023 kl. 21:54 skrev Bjørn Jørgensen <bjornjorgensen@gmail.com
>:
> percentile_approx returns the approximate percentile(s)
> <https://github.com/apache/spark/pull/14868> The memory consumption is
> bounded. The larger accuracy parameter we choose, the smaller error we get.
> The default accuracy value is 10000, to match with Hive default setting.
> Choose a smaller value for a smaller memory footprint.
>
> When I run my code on a single PC where N = 10 millions it takes 22.52
> seconds. Notebook added.
>
> I don't think that the question asker will have only returned the top
> 20 percentages.
>
>
> lør. 16. sep. 2023 kl. 17:49 skrev Mich Talebzadeh <
> mich.talebzadeh@gmail.com>:
>
>> Hi Bjorn,
>>
>> I thought that one is better off using percentile_approx as it seems to
>> be the recommended approach for computing percentiles and can simplify the
>> code.
>> I have modified your code to use percentile_approx rather than manually
>> computing it. It would be interesting to hear ideas on this.
>>
>> Here is the code:
>>
>> # Standard library imports
>> import json
>> import multiprocessing
>> import os
>> import re
>> import sys
>> import random
>>
>> # Third-party imports
>> import numpy as np
>> import pandas as pd
>> import pyarrow
>>
>> # Pyspark imports
>> from pyspark import SparkConf, SparkContext
>> from pyspark.sql import SparkSession, functions as F, Window
>> from pyspark.sql.functions import (
>> col, concat, concat_ws, expr, lit, trim, udf
>> )
>> from pyspark.sql.types import (
>> IntegerType, StringType, StructField, StructType,
>> DoubleType, TimestampType
>> )
>> from pyspark import pandas as ps
>>
>> os.environ["PYARROW_IGNORE_TIMEZONE"] = "1"
>>
>> number_cores = int(multiprocessing.cpu_count())
>>
>> mem_bytes = os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES") #
>> e.g. 4015976448
>> memory_gb = int(mem_bytes / (1024.0**3)) # e.g. 3.74
>>
>>
>> def get_spark_session(app_name: str, conf: SparkConf):
>> conf.setMaster("local[{}]".format(number_cores))
>> conf.set("spark.driver.memory", "{}g".format(memory_gb)).set(
>> "spark.sql.repl.eagerEval.enabled", "True"
>> ).set("spark.sql.adaptive.enabled", "True").set(
>> "spark.serializer", "org.apache.spark.serializer.KryoSerializer"
>> ).set(
>> "spark.sql.repl.eagerEval.maxNumRows", "10000"
>> )
>>
>> return
>> SparkSession.builder.appName(app_name).config(conf=conf).getOrCreate()
>>
>>
>> spark = get_spark_session("My super app", SparkConf())
>> sc = SparkContext.getOrCreate()
>> sc.setLogLevel("ERROR")
>>
>> def generate_ip():
>> return ".".join(str(random.randint(0, 255)) for _ in range(4))
>>
>> def generate_timestamp():
>> return pd.Timestamp(
>> year=random.randint(2021, 2023),
>> month=random.randint(1, 12),
>> day=random.randint(1, 28),
>> hour=random.randint(0, 23),
>> minute=random.randint(0, 59),
>> second=random.randint(0, 59)
>> )
>>
>> def random_gbps():
>> return random.uniform(0, 10)
>>
>> # Number of rows
>> n = 20
>>
>> data = [{"incoming_ips": generate_ip(), "gbps": random_gbps(),
>> "date_time": generate_timestamp()} for _ in range(n)]
>> df = spark.createDataFrame(pd.DataFrame(data))
>> df.show()
>>
>> agg_df = df.groupBy("incoming_ips").agg(F.sum("gbps").alias("total_gbps"))
>>
>> windowRank = Window.orderBy(F.col("total_gbps").desc())
>> agg_df = agg_df.withColumn("rank", F.percent_rank().over(windowRank))
>>
>> # Calculate the 80th percentile value
>> percentile_80 = agg_df.agg(F.expr("percentile_approx(total_gbps,
>> 0.8)").alias("percentile_80")).collect()[0]["percentile_80"]
>>
>> # Filter the DataFrame based on the condition
>> filtered_df = df.filter(df["gbps"] >= percentile_80)
>>
>> # Show the filtered DataFrame
>> print(f"Filtered DataFrame")
>> filtered_df.show()
>>
>> print(f"Total rows in data frame = {df.count()}")
>> print(f"Result satisfying 80% percentile = {filtered_df.count()}")
>>
>> And this is the results
>>
>> +---------------+------------------+-------------------+
>> | incoming_ips| gbps| date_time|
>> +---------------+------------------+-------------------+
>> |129.189.130.141|2.6517421918102335|2021-09-06 08:29:25|
>> | 215.177.39.239|1.8210013026429361|2023-10-10 17:00:13|
>> | 78.202.71.184| 8.060958370556456|2022-02-22 04:25:03|
>> |219.100.198.137|0.3449002002472945|2023-09-28 01:39:44|
>> |234.234.156.107|2.6187481766507013|2022-11-16 11:33:41|
>> | 6.223.135.194|0.3510752223686242|2022-01-24 04:13:53|
>> | 147.118.171.59|6.4071750880652765|2023-10-08 16:49:10|
>> | 75.41.101.165|2.1484984272041685|2022-07-13 21:02:58|
>> | 163.26.238.22| 9.8999646499433|2023-01-12 17:54:44|
>> | 184.145.98.231|1.8875849709728088|2022-09-18 19:53:58|
>> | 125.77.236.177| 1.17126350326476|2021-08-19 18:48:42|
>> | 34.103.211.39| 9.51081430594299|2023-02-05 18:39:23|
>> | 117.37.42.91| 1.122437784309721|2021-03-23 17:27:27|
>> | 108.115.42.171| 8.165187506266607|2023-07-26 03:57:50|
>> | 98.105.153.129| 9.284242190156004|2023-10-10 22:36:47|
>> | 145.35.252.142| 9.787384042283957|2022-08-26 00:53:27|
>> | 18.76.138.108| 6.939770760444909|2022-04-01 01:18:27|
>> | 31.33.71.26| 4.820947188427366|2021-06-10 22:02:51|
>> | 135.22.8.38| 9.587849542001745|2021-09-21 15:11:59|
>> |104.231.110.207| 9.045897927807715|2023-06-28 06:01:00|
>> +---------------+------------------+-------------------+
>>
>> Filtered DataFrame
>> +--------------+-----------------+-------------------+
>> | incoming_ips| gbps| date_time|
>> +--------------+-----------------+-------------------+
>> | 163.26.238.22| 9.8999646499433|2023-01-12 17:54:44|
>> | 34.103.211.39| 9.51081430594299|2023-02-05 18:39:23|
>> |98.105.153.129|9.284242190156004|2023-10-10 22:36:47|
>> |145.35.252.142|9.787384042283957|2022-08-26 00:53:27|
>> | 135.22.8.38|9.587849542001745|2021-09-21 15:11:59|
>> +--------------+-----------------+-------------------+
>>
>> Total rows in data frame = 20
>> Result satisfying 80% percentile = 5
>>
>> Cheers
>> Mich Talebzadeh,
>> Distinguished Technologist, Solutions Architect & Engineer
>> London
>> United Kingdom
>>
>>
>> view my Linkedin profile
>> <https://www.linkedin.com/in/mich-talebzadeh-ph-d-5205b2/>
>>
>>
>> https://en.everybodywiki.com/Mich_Talebzadeh
>>
>>
>>
>> *Disclaimer:* Use it at your own risk. Any and all responsibility for
>> any loss, damage or destruction of data or any other property which may
>> arise from relying on this email's technical content is explicitly
>> disclaimed. The author will in no case be liable for any monetary damages
>> arising from such loss, damage or destruction.
>>
>>
>>
>>
>> On Sat, 16 Sept 2023 at 11:46, Mich Talebzadeh <mi...@gmail.com>
>> wrote:
>>
>>> Happy Saturday coding 😁
>>>
>>>
>>> Mich Talebzadeh,
>>> Distinguished Technologist, Solutions Architect & Engineer
>>> London
>>> United Kingdom
>>>
>>>
>>> view my Linkedin profile
>>> <https://www.linkedin.com/in/mich-talebzadeh-ph-d-5205b2/>
>>>
>>>
>>> https://en.everybodywiki.com/Mich_Talebzadeh
>>>
>>>
>>>
>>> *Disclaimer:* Use it at your own risk. Any and all responsibility for
>>> any loss, damage or destruction of data or any other property which may
>>> arise from relying on this email's technical content is explicitly
>>> disclaimed. The author will in no case be liable for any monetary damages
>>> arising from such loss, damage or destruction.
>>>
>>>
>>>
>>>
>>> On Sat, 16 Sept 2023 at 11:30, Bjørn Jørgensen <bj...@gmail.com>
>>> wrote:
>>>
>>>> ah.. yes that's right.
>>>> I did have to use some time on this one and I was having some issues
>>>> with the code.
>>>> I restart the notebook kernel now and rerun it and I get the same
>>>> result.
>>>>
>>>> lør. 16. sep. 2023 kl. 11:41 skrev Mich Talebzadeh <
>>>> mich.talebzadeh@gmail.com>:
>>>>
>>>>> Splendid code. A minor error glancing at your code.
>>>>>
>>>>> print(df.count())
>>>>> print(result_df.count())
>>>>>
>>>>>
>>>>> You have not defined result_df. I gather you meant "result"?
>>>>>
>>>>>
>>>>> print(result.count())
>>>>>
>>>>>
>>>>> That should fix it 🤔
>>>>>
>>>>> HTH
>>>>>
>>>>>
>>>>> Mich Talebzadeh,
>>>>> Distinguished Technologist, Solutions Architect & Engineer
>>>>> London
>>>>> United Kingdom
>>>>>
>>>>>
>>>>> view my Linkedin profile
>>>>> <https://www.linkedin.com/in/mich-talebzadeh-ph-d-5205b2/>
>>>>>
>>>>>
>>>>> https://en.everybodywiki.com/Mich_Talebzadeh
>>>>>
>>>>>
>>>>>
>>>>> *Disclaimer:* Use it at your own risk. Any and all responsibility for
>>>>> any loss, damage or destruction of data or any other property which may
>>>>> arise from relying on this email's technical content is explicitly
>>>>> disclaimed. The author will in no case be liable for any monetary damages
>>>>> arising from such loss, damage or destruction.
>>>>>
>>>>>
>>>>>
>>>>>
>>>>> On Sat, 16 Sept 2023 at 06:00, Bjørn Jørgensen <
>>>>> bjornjorgensen@gmail.com> wrote:
>>>>>
>>>>>> Something like this?
>>>>>>
>>>>>>
>>>>>> # Standard library imports
>>>>>> import json
>>>>>> import multiprocessing
>>>>>> import os
>>>>>> import re
>>>>>> import sys
>>>>>> import random
>>>>>>
>>>>>> # Third-party imports
>>>>>> import numpy as np
>>>>>> import pandas as pd
>>>>>> import pyarrow
>>>>>>
>>>>>> # Pyspark imports
>>>>>> from pyspark import SparkConf, SparkContext
>>>>>> from pyspark.sql import SparkSession, functions as F, Window
>>>>>> from pyspark.sql.functions import (
>>>>>> col, concat, concat_ws, expr, lit, trim, udf
>>>>>> )
>>>>>> from pyspark.sql.types import (
>>>>>> IntegerType, StringType, StructField, StructType,
>>>>>> DoubleType, TimestampType
>>>>>> )
>>>>>> from pyspark import pandas as ps
>>>>>>
>>>>>> os.environ["PYARROW_IGNORE_TIMEZONE"] = "1"
>>>>>>
>>>>>> number_cores = int(multiprocessing.cpu_count())
>>>>>>
>>>>>> mem_bytes = os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES") # e.g. 4015976448
>>>>>> memory_gb = int(mem_bytes / (1024.0**3)) # e.g. 3.74
>>>>>>
>>>>>>
>>>>>> def get_spark_session(app_name: str, conf: SparkConf):
>>>>>> conf.setMaster("local[{}]".format(number_cores))
>>>>>> conf.set("spark.driver.memory", "{}g".format(memory_gb)).set(
>>>>>> "spark.sql.repl.eagerEval.enabled", "True"
>>>>>> ).set("spark.sql.adaptive.enabled", "True").set(
>>>>>> "spark.serializer", "org.apache.spark.serializer.KryoSerializer"
>>>>>> ).set(
>>>>>> "spark.sql.repl.eagerEval.maxNumRows", "10000"
>>>>>> ).set(
>>>>>> "sc.setLogLevel", "ERROR"
>>>>>> )
>>>>>>
>>>>>> return SparkSession.builder.appName(app_name).config(conf=conf).getOrCreate()
>>>>>>
>>>>>>
>>>>>> spark = get_spark_session("My super app", SparkConf())
>>>>>> spark.sparkContext.setLogLevel("ERROR")
>>>>>>
>>>>>>
>>>>>>
>>>>>> def generate_ip():
>>>>>> return ".".join(str(random.randint(0, 255)) for _ in range(4))
>>>>>>
>>>>>> def generate_timestamp():
>>>>>> return pd.Timestamp(
>>>>>> year=random.randint(2021, 2023),
>>>>>> month=random.randint(1, 12),
>>>>>> day=random.randint(1, 28),
>>>>>> hour=random.randint(0, 23),
>>>>>> minute=random.randint(0, 59),
>>>>>> second=random.randint(0, 59)
>>>>>> )
>>>>>>
>>>>>> def random_gbps():
>>>>>> return random.uniform(0, 10)
>>>>>>
>>>>>> # Number of rows
>>>>>> n = 20
>>>>>>
>>>>>> data = [{"incoming_ips": generate_ip(), "gbps": random_gbps(), "date_time": generate_timestamp()} for _ in range(n)]
>>>>>> df = spark.createDataFrame(pd.DataFrame(data))
>>>>>> df.show()
>>>>>>
>>>>>> agg_df = df.groupBy("incoming_ips").agg(F.sum("gbps").alias("total_gbps"))
>>>>>>
>>>>>> windowRank = Window.orderBy(F.col("total_gbps").desc())
>>>>>> agg_df = agg_df.withColumn("rank", F.percent_rank().over(windowRank))
>>>>>>
>>>>>> top_80_ips = agg_df.filter(F.col("rank") <= 0.80)
>>>>>> result = df.join(top_80_ips, on="incoming_ips", how="inner").select("incoming_ips", "gbps", "date_time")
>>>>>> result.show()
>>>>>>
>>>>>> print(df.count())
>>>>>> print(result_df.count())
>>>>>>
>>>>>>
>>>>>> +---------------+-------------------+-------------------+
>>>>>> | incoming_ips| gbps| date_time|
>>>>>> +---------------+-------------------+-------------------+
>>>>>> | 66.186.8.130| 5.074283124722104|2022-03-12 05:09:16|
>>>>>> | 155.45.76.235| 0.6736194760917324|2021-06-19 03:36:28|
>>>>>> | 237.51.137.200|0.43334812775057685|2022-04-27 08:08:47|
>>>>>> | 78.4.48.171| 7.5675453578753435|2022-08-21 18:55:48|
>>>>>> | 241.84.163.17| 3.5681655964070815|2021-01-24 20:39:50|
>>>>>> |130.255.202.138| 6.066112278135983|2023-07-07 22:26:15|
>>>>>> | 198.33.206.140| 1.9147905257021836|2023-03-01 04:44:14|
>>>>>> | 84.183.253.20| 7.707176860385722|2021-08-26 23:24:31|
>>>>>> |218.163.165.232| 9.458673015973213|2021-02-22 12:13:15|
>>>>>> | 62.57.20.153| 1.5764916247359229|2021-11-06 12:41:59|
>>>>>> | 245.24.168.152|0.07452805411698016|2021-06-04 16:14:36|
>>>>>> | 98.171.202.249| 3.546118349483626|2022-07-05 10:55:26|
>>>>>> | 210.5.246.85|0.02430730260109759|2022-04-08 17:26:04|
>>>>>> | 13.236.170.177| 2.41361938344535|2021-08-11 02:19:06|
>>>>>> |180.140.248.193| 0.9512956363005021|2021-06-27 18:16:58|
>>>>>> | 26.140.88.127| 7.51335778127692|2023-06-02 14:13:30|
>>>>>> | 7.118.207.252| 6.450499049816286|2022-12-11 06:36:20|
>>>>>> | 11.8.10.136| 8.750329246667354|2023-02-03 05:33:16|
>>>>>> | 232.140.56.86| 4.289740988237201|2023-02-22 20:10:09|
>>>>>> | 68.117.9.255| 5.384340363304169|2022-12-03 09:55:26|
>>>>>> +---------------+-------------------+-------------------+
>>>>>>
>>>>>> +---------------+------------------+-------------------+
>>>>>> | incoming_ips| gbps| date_time|
>>>>>> +---------------+------------------+-------------------+
>>>>>> | 66.186.8.130| 5.074283124722104|2022-03-12 05:09:16|
>>>>>> | 241.84.163.17|3.5681655964070815|2021-01-24 20:39:50|
>>>>>> | 78.4.48.171|7.5675453578753435|2022-08-21 18:55:48|
>>>>>> |130.255.202.138| 6.066112278135983|2023-07-07 22:26:15|
>>>>>> | 198.33.206.140|1.9147905257021836|2023-03-01 04:44:14|
>>>>>> | 84.183.253.20| 7.707176860385722|2021-08-26 23:24:31|
>>>>>> |218.163.165.232| 9.458673015973213|2021-02-22 12:13:15|
>>>>>> | 62.57.20.153|1.5764916247359229|2021-11-06 12:41:59|
>>>>>> | 98.171.202.249| 3.546118349483626|2022-07-05 10:55:26|
>>>>>> |180.140.248.193|0.9512956363005021|2021-06-27 18:16:58|
>>>>>> | 13.236.170.177| 2.41361938344535|2021-08-11 02:19:06|
>>>>>> | 26.140.88.127| 7.51335778127692|2023-06-02 14:13:30|
>>>>>> | 7.118.207.252| 6.450499049816286|2022-12-11 06:36:20|
>>>>>> | 11.8.10.136| 8.750329246667354|2023-02-03 05:33:16|
>>>>>> | 232.140.56.86| 4.289740988237201|2023-02-22 20:10:09|
>>>>>> | 68.117.9.255| 5.384340363304169|2022-12-03 09:55:26|
>>>>>> +---------------+------------------+-------------------+
>>>>>>
>>>>>> 20
>>>>>> 16
>>>>>>
>>>>>>
>>>>>>
>>>>>> fre. 15. sep. 2023 kl. 20:14 skrev ashok34668@yahoo.com.INVALID
>>>>>> <as...@yahoo.com.invalid>:
>>>>>>
>>>>>>> Hi team,
>>>>>>>
>>>>>>> I am using PySpark 3.4
>>>>>>>
>>>>>>> I have a table of million rows that has few columns. among them
>>>>>>> incoming ips and what is known as gbps (Gigabytes per second) and
>>>>>>> date and time of incoming ip.
>>>>>>>
>>>>>>> I want to filter out 20% of low active ips and work on the remainder
>>>>>>> of data. How can I do thiis in PySpark?
>>>>>>>
>>>>>>> Thanks
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>
>>>>>> --
>>>>>> Bjørn Jørgensen
>>>>>> Vestre Aspehaug 4, 6010 Ålesund
>>>>>> Norge
>>>>>>
>>>>>> +47 480 94 297
>>>>>>
>>>>>
>>>>
>>>> --
>>>> Bjørn Jørgensen
>>>> Vestre Aspehaug 4, 6010 Ålesund
>>>> Norge
>>>>
>>>> +47 480 94 297
>>>>
>>>
>
> --
> Bjørn Jørgensen
> Vestre Aspehaug 4, 6010 Ålesund
> Norge
>
> +47 480 94 297
>
--
Bjørn Jørgensen
Vestre Aspehaug 4, 6010 Ålesund
Norge
+47 480 94 297
Re: Filter out 20% of rows
Posted by Bjørn Jørgensen <bj...@gmail.com>.
percentile_approx returns the approximate percentile(s)
<https://github.com/apache/spark/pull/14868> The memory consumption is
bounded. The larger accuracy parameter we choose, the smaller error we get.
The default accuracy value is 10000, to match with Hive default setting.
Choose a smaller value for a smaller memory footprint.
When I run my code on a single PC where N = 10 millions it takes 22.52
seconds. Notebook added.
I don't think that the question asker will have only returned the top
20 percentages.
lør. 16. sep. 2023 kl. 17:49 skrev Mich Talebzadeh <
mich.talebzadeh@gmail.com>:
> Hi Bjorn,
>
> I thought that one is better off using percentile_approx as it seems to be
> the recommended approach for computing percentiles and can simplify the
> code.
> I have modified your code to use percentile_approx rather than manually
> computing it. It would be interesting to hear ideas on this.
>
> Here is the code:
>
> # Standard library imports
> import json
> import multiprocessing
> import os
> import re
> import sys
> import random
>
> # Third-party imports
> import numpy as np
> import pandas as pd
> import pyarrow
>
> # Pyspark imports
> from pyspark import SparkConf, SparkContext
> from pyspark.sql import SparkSession, functions as F, Window
> from pyspark.sql.functions import (
> col, concat, concat_ws, expr, lit, trim, udf
> )
> from pyspark.sql.types import (
> IntegerType, StringType, StructField, StructType,
> DoubleType, TimestampType
> )
> from pyspark import pandas as ps
>
> os.environ["PYARROW_IGNORE_TIMEZONE"] = "1"
>
> number_cores = int(multiprocessing.cpu_count())
>
> mem_bytes = os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES") #
> e.g. 4015976448
> memory_gb = int(mem_bytes / (1024.0**3)) # e.g. 3.74
>
>
> def get_spark_session(app_name: str, conf: SparkConf):
> conf.setMaster("local[{}]".format(number_cores))
> conf.set("spark.driver.memory", "{}g".format(memory_gb)).set(
> "spark.sql.repl.eagerEval.enabled", "True"
> ).set("spark.sql.adaptive.enabled", "True").set(
> "spark.serializer", "org.apache.spark.serializer.KryoSerializer"
> ).set(
> "spark.sql.repl.eagerEval.maxNumRows", "10000"
> )
>
> return
> SparkSession.builder.appName(app_name).config(conf=conf).getOrCreate()
>
>
> spark = get_spark_session("My super app", SparkConf())
> sc = SparkContext.getOrCreate()
> sc.setLogLevel("ERROR")
>
> def generate_ip():
> return ".".join(str(random.randint(0, 255)) for _ in range(4))
>
> def generate_timestamp():
> return pd.Timestamp(
> year=random.randint(2021, 2023),
> month=random.randint(1, 12),
> day=random.randint(1, 28),
> hour=random.randint(0, 23),
> minute=random.randint(0, 59),
> second=random.randint(0, 59)
> )
>
> def random_gbps():
> return random.uniform(0, 10)
>
> # Number of rows
> n = 20
>
> data = [{"incoming_ips": generate_ip(), "gbps": random_gbps(),
> "date_time": generate_timestamp()} for _ in range(n)]
> df = spark.createDataFrame(pd.DataFrame(data))
> df.show()
>
> agg_df = df.groupBy("incoming_ips").agg(F.sum("gbps").alias("total_gbps"))
>
> windowRank = Window.orderBy(F.col("total_gbps").desc())
> agg_df = agg_df.withColumn("rank", F.percent_rank().over(windowRank))
>
> # Calculate the 80th percentile value
> percentile_80 = agg_df.agg(F.expr("percentile_approx(total_gbps,
> 0.8)").alias("percentile_80")).collect()[0]["percentile_80"]
>
> # Filter the DataFrame based on the condition
> filtered_df = df.filter(df["gbps"] >= percentile_80)
>
> # Show the filtered DataFrame
> print(f"Filtered DataFrame")
> filtered_df.show()
>
> print(f"Total rows in data frame = {df.count()}")
> print(f"Result satisfying 80% percentile = {filtered_df.count()}")
>
> And this is the results
>
> +---------------+------------------+-------------------+
> | incoming_ips| gbps| date_time|
> +---------------+------------------+-------------------+
> |129.189.130.141|2.6517421918102335|2021-09-06 08:29:25|
> | 215.177.39.239|1.8210013026429361|2023-10-10 17:00:13|
> | 78.202.71.184| 8.060958370556456|2022-02-22 04:25:03|
> |219.100.198.137|0.3449002002472945|2023-09-28 01:39:44|
> |234.234.156.107|2.6187481766507013|2022-11-16 11:33:41|
> | 6.223.135.194|0.3510752223686242|2022-01-24 04:13:53|
> | 147.118.171.59|6.4071750880652765|2023-10-08 16:49:10|
> | 75.41.101.165|2.1484984272041685|2022-07-13 21:02:58|
> | 163.26.238.22| 9.8999646499433|2023-01-12 17:54:44|
> | 184.145.98.231|1.8875849709728088|2022-09-18 19:53:58|
> | 125.77.236.177| 1.17126350326476|2021-08-19 18:48:42|
> | 34.103.211.39| 9.51081430594299|2023-02-05 18:39:23|
> | 117.37.42.91| 1.122437784309721|2021-03-23 17:27:27|
> | 108.115.42.171| 8.165187506266607|2023-07-26 03:57:50|
> | 98.105.153.129| 9.284242190156004|2023-10-10 22:36:47|
> | 145.35.252.142| 9.787384042283957|2022-08-26 00:53:27|
> | 18.76.138.108| 6.939770760444909|2022-04-01 01:18:27|
> | 31.33.71.26| 4.820947188427366|2021-06-10 22:02:51|
> | 135.22.8.38| 9.587849542001745|2021-09-21 15:11:59|
> |104.231.110.207| 9.045897927807715|2023-06-28 06:01:00|
> +---------------+------------------+-------------------+
>
> Filtered DataFrame
> +--------------+-----------------+-------------------+
> | incoming_ips| gbps| date_time|
> +--------------+-----------------+-------------------+
> | 163.26.238.22| 9.8999646499433|2023-01-12 17:54:44|
> | 34.103.211.39| 9.51081430594299|2023-02-05 18:39:23|
> |98.105.153.129|9.284242190156004|2023-10-10 22:36:47|
> |145.35.252.142|9.787384042283957|2022-08-26 00:53:27|
> | 135.22.8.38|9.587849542001745|2021-09-21 15:11:59|
> +--------------+-----------------+-------------------+
>
> Total rows in data frame = 20
> Result satisfying 80% percentile = 5
>
> Cheers
> Mich Talebzadeh,
> Distinguished Technologist, Solutions Architect & Engineer
> London
> United Kingdom
>
>
> view my Linkedin profile
> <https://www.linkedin.com/in/mich-talebzadeh-ph-d-5205b2/>
>
>
> https://en.everybodywiki.com/Mich_Talebzadeh
>
>
>
> *Disclaimer:* Use it at your own risk. Any and all responsibility for any
> loss, damage or destruction of data or any other property which may arise
> from relying on this email's technical content is explicitly disclaimed.
> The author will in no case be liable for any monetary damages arising from
> such loss, damage or destruction.
>
>
>
>
> On Sat, 16 Sept 2023 at 11:46, Mich Talebzadeh <mi...@gmail.com>
> wrote:
>
>> Happy Saturday coding 😁
>>
>>
>> Mich Talebzadeh,
>> Distinguished Technologist, Solutions Architect & Engineer
>> London
>> United Kingdom
>>
>>
>> view my Linkedin profile
>> <https://www.linkedin.com/in/mich-talebzadeh-ph-d-5205b2/>
>>
>>
>> https://en.everybodywiki.com/Mich_Talebzadeh
>>
>>
>>
>> *Disclaimer:* Use it at your own risk. Any and all responsibility for
>> any loss, damage or destruction of data or any other property which may
>> arise from relying on this email's technical content is explicitly
>> disclaimed. The author will in no case be liable for any monetary damages
>> arising from such loss, damage or destruction.
>>
>>
>>
>>
>> On Sat, 16 Sept 2023 at 11:30, Bjørn Jørgensen <bj...@gmail.com>
>> wrote:
>>
>>> ah.. yes that's right.
>>> I did have to use some time on this one and I was having some issues
>>> with the code.
>>> I restart the notebook kernel now and rerun it and I get the same
>>> result.
>>>
>>> lør. 16. sep. 2023 kl. 11:41 skrev Mich Talebzadeh <
>>> mich.talebzadeh@gmail.com>:
>>>
>>>> Splendid code. A minor error glancing at your code.
>>>>
>>>> print(df.count())
>>>> print(result_df.count())
>>>>
>>>>
>>>> You have not defined result_df. I gather you meant "result"?
>>>>
>>>>
>>>> print(result.count())
>>>>
>>>>
>>>> That should fix it 🤔
>>>>
>>>> HTH
>>>>
>>>>
>>>> Mich Talebzadeh,
>>>> Distinguished Technologist, Solutions Architect & Engineer
>>>> London
>>>> United Kingdom
>>>>
>>>>
>>>> view my Linkedin profile
>>>> <https://www.linkedin.com/in/mich-talebzadeh-ph-d-5205b2/>
>>>>
>>>>
>>>> https://en.everybodywiki.com/Mich_Talebzadeh
>>>>
>>>>
>>>>
>>>> *Disclaimer:* Use it at your own risk. Any and all responsibility for
>>>> any loss, damage or destruction of data or any other property which may
>>>> arise from relying on this email's technical content is explicitly
>>>> disclaimed. The author will in no case be liable for any monetary damages
>>>> arising from such loss, damage or destruction.
>>>>
>>>>
>>>>
>>>>
>>>> On Sat, 16 Sept 2023 at 06:00, Bjørn Jørgensen <
>>>> bjornjorgensen@gmail.com> wrote:
>>>>
>>>>> Something like this?
>>>>>
>>>>>
>>>>> # Standard library imports
>>>>> import json
>>>>> import multiprocessing
>>>>> import os
>>>>> import re
>>>>> import sys
>>>>> import random
>>>>>
>>>>> # Third-party imports
>>>>> import numpy as np
>>>>> import pandas as pd
>>>>> import pyarrow
>>>>>
>>>>> # Pyspark imports
>>>>> from pyspark import SparkConf, SparkContext
>>>>> from pyspark.sql import SparkSession, functions as F, Window
>>>>> from pyspark.sql.functions import (
>>>>> col, concat, concat_ws, expr, lit, trim, udf
>>>>> )
>>>>> from pyspark.sql.types import (
>>>>> IntegerType, StringType, StructField, StructType,
>>>>> DoubleType, TimestampType
>>>>> )
>>>>> from pyspark import pandas as ps
>>>>>
>>>>> os.environ["PYARROW_IGNORE_TIMEZONE"] = "1"
>>>>>
>>>>> number_cores = int(multiprocessing.cpu_count())
>>>>>
>>>>> mem_bytes = os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES") # e.g. 4015976448
>>>>> memory_gb = int(mem_bytes / (1024.0**3)) # e.g. 3.74
>>>>>
>>>>>
>>>>> def get_spark_session(app_name: str, conf: SparkConf):
>>>>> conf.setMaster("local[{}]".format(number_cores))
>>>>> conf.set("spark.driver.memory", "{}g".format(memory_gb)).set(
>>>>> "spark.sql.repl.eagerEval.enabled", "True"
>>>>> ).set("spark.sql.adaptive.enabled", "True").set(
>>>>> "spark.serializer", "org.apache.spark.serializer.KryoSerializer"
>>>>> ).set(
>>>>> "spark.sql.repl.eagerEval.maxNumRows", "10000"
>>>>> ).set(
>>>>> "sc.setLogLevel", "ERROR"
>>>>> )
>>>>>
>>>>> return SparkSession.builder.appName(app_name).config(conf=conf).getOrCreate()
>>>>>
>>>>>
>>>>> spark = get_spark_session("My super app", SparkConf())
>>>>> spark.sparkContext.setLogLevel("ERROR")
>>>>>
>>>>>
>>>>>
>>>>> def generate_ip():
>>>>> return ".".join(str(random.randint(0, 255)) for _ in range(4))
>>>>>
>>>>> def generate_timestamp():
>>>>> return pd.Timestamp(
>>>>> year=random.randint(2021, 2023),
>>>>> month=random.randint(1, 12),
>>>>> day=random.randint(1, 28),
>>>>> hour=random.randint(0, 23),
>>>>> minute=random.randint(0, 59),
>>>>> second=random.randint(0, 59)
>>>>> )
>>>>>
>>>>> def random_gbps():
>>>>> return random.uniform(0, 10)
>>>>>
>>>>> # Number of rows
>>>>> n = 20
>>>>>
>>>>> data = [{"incoming_ips": generate_ip(), "gbps": random_gbps(), "date_time": generate_timestamp()} for _ in range(n)]
>>>>> df = spark.createDataFrame(pd.DataFrame(data))
>>>>> df.show()
>>>>>
>>>>> agg_df = df.groupBy("incoming_ips").agg(F.sum("gbps").alias("total_gbps"))
>>>>>
>>>>> windowRank = Window.orderBy(F.col("total_gbps").desc())
>>>>> agg_df = agg_df.withColumn("rank", F.percent_rank().over(windowRank))
>>>>>
>>>>> top_80_ips = agg_df.filter(F.col("rank") <= 0.80)
>>>>> result = df.join(top_80_ips, on="incoming_ips", how="inner").select("incoming_ips", "gbps", "date_time")
>>>>> result.show()
>>>>>
>>>>> print(df.count())
>>>>> print(result_df.count())
>>>>>
>>>>>
>>>>> +---------------+-------------------+-------------------+
>>>>> | incoming_ips| gbps| date_time|
>>>>> +---------------+-------------------+-------------------+
>>>>> | 66.186.8.130| 5.074283124722104|2022-03-12 05:09:16|
>>>>> | 155.45.76.235| 0.6736194760917324|2021-06-19 03:36:28|
>>>>> | 237.51.137.200|0.43334812775057685|2022-04-27 08:08:47|
>>>>> | 78.4.48.171| 7.5675453578753435|2022-08-21 18:55:48|
>>>>> | 241.84.163.17| 3.5681655964070815|2021-01-24 20:39:50|
>>>>> |130.255.202.138| 6.066112278135983|2023-07-07 22:26:15|
>>>>> | 198.33.206.140| 1.9147905257021836|2023-03-01 04:44:14|
>>>>> | 84.183.253.20| 7.707176860385722|2021-08-26 23:24:31|
>>>>> |218.163.165.232| 9.458673015973213|2021-02-22 12:13:15|
>>>>> | 62.57.20.153| 1.5764916247359229|2021-11-06 12:41:59|
>>>>> | 245.24.168.152|0.07452805411698016|2021-06-04 16:14:36|
>>>>> | 98.171.202.249| 3.546118349483626|2022-07-05 10:55:26|
>>>>> | 210.5.246.85|0.02430730260109759|2022-04-08 17:26:04|
>>>>> | 13.236.170.177| 2.41361938344535|2021-08-11 02:19:06|
>>>>> |180.140.248.193| 0.9512956363005021|2021-06-27 18:16:58|
>>>>> | 26.140.88.127| 7.51335778127692|2023-06-02 14:13:30|
>>>>> | 7.118.207.252| 6.450499049816286|2022-12-11 06:36:20|
>>>>> | 11.8.10.136| 8.750329246667354|2023-02-03 05:33:16|
>>>>> | 232.140.56.86| 4.289740988237201|2023-02-22 20:10:09|
>>>>> | 68.117.9.255| 5.384340363304169|2022-12-03 09:55:26|
>>>>> +---------------+-------------------+-------------------+
>>>>>
>>>>> +---------------+------------------+-------------------+
>>>>> | incoming_ips| gbps| date_time|
>>>>> +---------------+------------------+-------------------+
>>>>> | 66.186.8.130| 5.074283124722104|2022-03-12 05:09:16|
>>>>> | 241.84.163.17|3.5681655964070815|2021-01-24 20:39:50|
>>>>> | 78.4.48.171|7.5675453578753435|2022-08-21 18:55:48|
>>>>> |130.255.202.138| 6.066112278135983|2023-07-07 22:26:15|
>>>>> | 198.33.206.140|1.9147905257021836|2023-03-01 04:44:14|
>>>>> | 84.183.253.20| 7.707176860385722|2021-08-26 23:24:31|
>>>>> |218.163.165.232| 9.458673015973213|2021-02-22 12:13:15|
>>>>> | 62.57.20.153|1.5764916247359229|2021-11-06 12:41:59|
>>>>> | 98.171.202.249| 3.546118349483626|2022-07-05 10:55:26|
>>>>> |180.140.248.193|0.9512956363005021|2021-06-27 18:16:58|
>>>>> | 13.236.170.177| 2.41361938344535|2021-08-11 02:19:06|
>>>>> | 26.140.88.127| 7.51335778127692|2023-06-02 14:13:30|
>>>>> | 7.118.207.252| 6.450499049816286|2022-12-11 06:36:20|
>>>>> | 11.8.10.136| 8.750329246667354|2023-02-03 05:33:16|
>>>>> | 232.140.56.86| 4.289740988237201|2023-02-22 20:10:09|
>>>>> | 68.117.9.255| 5.384340363304169|2022-12-03 09:55:26|
>>>>> +---------------+------------------+-------------------+
>>>>>
>>>>> 20
>>>>> 16
>>>>>
>>>>>
>>>>>
>>>>> fre. 15. sep. 2023 kl. 20:14 skrev ashok34668@yahoo.com.INVALID
>>>>> <as...@yahoo.com.invalid>:
>>>>>
>>>>>> Hi team,
>>>>>>
>>>>>> I am using PySpark 3.4
>>>>>>
>>>>>> I have a table of million rows that has few columns. among them
>>>>>> incoming ips and what is known as gbps (Gigabytes per second) and
>>>>>> date and time of incoming ip.
>>>>>>
>>>>>> I want to filter out 20% of low active ips and work on the remainder
>>>>>> of data. How can I do thiis in PySpark?
>>>>>>
>>>>>> Thanks
>>>>>>
>>>>>>
>>>>>>
>>>>>>
>>>>>>
>>>>>
>>>>> --
>>>>> Bjørn Jørgensen
>>>>> Vestre Aspehaug 4, 6010 Ålesund
>>>>> Norge
>>>>>
>>>>> +47 480 94 297
>>>>>
>>>>
>>>
>>> --
>>> Bjørn Jørgensen
>>> Vestre Aspehaug 4, 6010 Ålesund
>>> Norge
>>>
>>> +47 480 94 297
>>>
>>
--
Bjørn Jørgensen
Vestre Aspehaug 4, 6010 Ålesund
Norge
+47 480 94 297
Re: Filter out 20% of rows
Posted by "ashok34668@yahoo.com.INVALID" <as...@yahoo.com.INVALID>.
Thank you Bjorn and Mich.
Appreciated
Best
On Saturday, 16 September 2023 at 16:50:04 BST, Mich Talebzadeh <mi...@gmail.com> wrote:
Hi Bjorn,
I thought that one is better off using percentile_approx as it seems to be the recommended approach for computing percentiles and can simplify the code. I have modified your code to use percentile_approx rather than manually computing it. It would be interesting to hear ideas on this.
Here is the code:
# Standard library importsimport jsonimport multiprocessingimport osimport reimport sysimport random
# Third-party importsimport numpy as npimport pandas as pdimport pyarrow
# Pyspark importsfrom pyspark import SparkConf, SparkContextfrom pyspark.sql import SparkSession, functions as F, Windowfrom pyspark.sql.functions import ( col, concat, concat_ws, expr, lit, trim, udf)from pyspark.sql.types import ( IntegerType, StringType, StructField, StructType, DoubleType, TimestampType)from pyspark import pandas as ps
os.environ["PYARROW_IGNORE_TIMEZONE"] = "1"
number_cores = int(multiprocessing.cpu_count())
mem_bytes = os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES") # e.g. 4015976448memory_gb = int(mem_bytes / (1024.0**3)) # e.g. 3.74
def get_spark_session(app_name: str, conf: SparkConf): conf.setMaster("local[{}]".format(number_cores)) conf.set("spark.driver.memory", "{}g".format(memory_gb)).set( "spark.sql.repl.eagerEval.enabled", "True" ).set("spark.sql.adaptive.enabled", "True").set( "spark.serializer", "org.apache.spark.serializer.KryoSerializer" ).set( "spark.sql.repl.eagerEval.maxNumRows", "10000" )
return SparkSession.builder.appName(app_name).config(conf=conf).getOrCreate()
spark = get_spark_session("My super app", SparkConf())sc = SparkContext.getOrCreate()sc.setLogLevel("ERROR")
def generate_ip(): return ".".join(str(random.randint(0, 255)) for _ in range(4))
def generate_timestamp(): return pd.Timestamp( year=random.randint(2021, 2023), month=random.randint(1, 12), day=random.randint(1, 28), hour=random.randint(0, 23), minute=random.randint(0, 59), second=random.randint(0, 59) )
def random_gbps(): return random.uniform(0, 10)
# Number of rowsn = 20
data = [{"incoming_ips": generate_ip(), "gbps": random_gbps(), "date_time": generate_timestamp()} for _ in range(n)]df = spark.createDataFrame(pd.DataFrame(data))df.show()
agg_df = df.groupBy("incoming_ips").agg(F.sum("gbps").alias("total_gbps"))
windowRank = Window.orderBy(F.col("total_gbps").desc())agg_df = agg_df.withColumn("rank", F.percent_rank().over(windowRank))
# Calculate the 80th percentile valuepercentile_80 = agg_df.agg(F.expr("percentile_approx(total_gbps, 0.8)").alias("percentile_80")).collect()[0]["percentile_80"]
# Filter the DataFrame based on the conditionfiltered_df = df.filter(df["gbps"] >= percentile_80)
# Show the filtered DataFrameprint(f"Filtered DataFrame")filtered_df.show()
print(f"Total rows in data frame = {df.count()}")print(f"Result satisfying 80% percentile = {filtered_df.count()}")
And this is the results
+---------------+------------------+-------------------+| incoming_ips| gbps| date_time|+---------------+------------------+-------------------+|129.189.130.141|2.6517421918102335|2021-09-06 08:29:25|| 215.177.39.239|1.8210013026429361|2023-10-10 17:00:13|| 78.202.71.184| 8.060958370556456|2022-02-22 04:25:03||219.100.198.137|0.3449002002472945|2023-09-28 01:39:44||234.234.156.107|2.6187481766507013|2022-11-16 11:33:41|| 6.223.135.194|0.3510752223686242|2022-01-24 04:13:53|| 147.118.171.59|6.4071750880652765|2023-10-08 16:49:10|| 75.41.101.165|2.1484984272041685|2022-07-13 21:02:58|| 163.26.238.22| 9.8999646499433|2023-01-12 17:54:44|| 184.145.98.231|1.8875849709728088|2022-09-18 19:53:58|| 125.77.236.177| 1.17126350326476|2021-08-19 18:48:42|| 34.103.211.39| 9.51081430594299|2023-02-05 18:39:23|| 117.37.42.91| 1.122437784309721|2021-03-23 17:27:27|| 108.115.42.171| 8.165187506266607|2023-07-26 03:57:50|| 98.105.153.129| 9.284242190156004|2023-10-10 22:36:47|| 145.35.252.142| 9.787384042283957|2022-08-26 00:53:27|| 18.76.138.108| 6.939770760444909|2022-04-01 01:18:27|| 31.33.71.26| 4.820947188427366|2021-06-10 22:02:51|| 135.22.8.38| 9.587849542001745|2021-09-21 15:11:59||104.231.110.207| 9.045897927807715|2023-06-28 06:01:00|+---------------+------------------+-------------------+
Filtered DataFrame+--------------+-----------------+-------------------+| incoming_ips| gbps| date_time|+--------------+-----------------+-------------------+| 163.26.238.22| 9.8999646499433|2023-01-12 17:54:44|| 34.103.211.39| 9.51081430594299|2023-02-05 18:39:23||98.105.153.129|9.284242190156004|2023-10-10 22:36:47||145.35.252.142|9.787384042283957|2022-08-26 00:53:27|| 135.22.8.38|9.587849542001745|2021-09-21 15:11:59|+--------------+-----------------+-------------------+
Total rows in data frame = 20Result satisfying 80% percentile = 5
CheersMich Talebzadeh,Distinguished Technologist, Solutions Architect & Engineer
London
United Kingdom
view my Linkedin profile
https://en.everybodywiki.com/Mich_Talebzadeh
Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destructionof data or any other property which may arise from relying on this email's technical content is explicitly disclaimed.The author will in no case be liable for any monetary damages arising from suchloss, damage or destruction.
On Sat, 16 Sept 2023 at 11:46, Mich Talebzadeh <mi...@gmail.com> wrote:
Happy Saturday coding 😁
Mich Talebzadeh,Distinguished Technologist, Solutions Architect & Engineer
LondonUnited Kingdom
view my Linkedin profile
https://en.everybodywiki.com/Mich_Talebzadeh
Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destructionof data or any other property which may arise from relying on this email's technical content is explicitly disclaimed.The author will in no case be liable for any monetary damages arising from suchloss, damage or destruction.
On Sat, 16 Sept 2023 at 11:30, Bjørn Jørgensen <bj...@gmail.com> wrote:
ah.. yes that's right. I did have to use some time on this one and I was having some issues with the code. I restart the notebook kernel now and rerun it and I get the same result.
lør. 16. sep. 2023 kl. 11:41 skrev Mich Talebzadeh <mi...@gmail.com>:
Splendid code. A minor error glancing at your code.
print(df.count())
print(result_df.count())
You have not defined result_df. I gather you meant "result"?
print(result.count())
That should fix it 🤔HTH
Mich Talebzadeh,Distinguished Technologist, Solutions Architect & Engineer
London
United Kingdom
view my Linkedin profile
https://en.everybodywiki.com/Mich_Talebzadeh
Disclaimer: Use it at your own risk. Any and all responsibility for any loss, damage or destructionof data or any other property which may arise from relying on this email's technical content is explicitly disclaimed.The author will in no case be liable for any monetary damages arising from suchloss, damage or destruction.
On Sat, 16 Sept 2023 at 06:00, Bjørn Jørgensen <bj...@gmail.com> wrote:
Something like this?
# Standard library imports
import json
import multiprocessing
import os
import re
import sys
import random
# Third-party imports
import numpy as np
import pandas as pd
import pyarrow
# Pyspark imports
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession, functions as F, Window
from pyspark.sql.functions import (
col, concat, concat_ws, expr, lit, trim, udf
)
from pyspark.sql.types import (
IntegerType, StringType, StructField, StructType,
DoubleType, TimestampType
)
from pyspark import pandas as ps
os.environ["PYARROW_IGNORE_TIMEZONE"] = "1"
number_cores = int(multiprocessing.cpu_count())
mem_bytes = os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES") # e.g. 4015976448
memory_gb = int(mem_bytes / (1024.0**3)) # e.g. 3.74
def get_spark_session(app_name: str, conf: SparkConf):
conf.setMaster("local[{}]".format(number_cores))
conf.set("spark.driver.memory", "{}g".format(memory_gb)).set(
"spark.sql.repl.eagerEval.enabled", "True"
).set("spark.sql.adaptive.enabled", "True").set(
"spark.serializer", "org.apache.spark.serializer.KryoSerializer"
).set(
"spark.sql.repl.eagerEval.maxNumRows", "10000"
).set(
"sc.setLogLevel", "ERROR"
)
return SparkSession.builder.appName(app_name).config(conf=conf).getOrCreate()
spark = get_spark_session("My super app", SparkConf())
spark.sparkContext.setLogLevel("ERROR")
def generate_ip():
return ".".join(str(random.randint(0, 255)) for _ in range(4))
def generate_timestamp():
return pd.Timestamp(
year=random.randint(2021, 2023),
month=random.randint(1, 12),
day=random.randint(1, 28),
hour=random.randint(0, 23),
minute=random.randint(0, 59),
second=random.randint(0, 59)
)
def random_gbps():
return random.uniform(0, 10)
# Number of rows
n = 20
data = [{"incoming_ips": generate_ip(), "gbps": random_gbps(), "date_time": generate_timestamp()} for _ in range(n)]
df = spark.createDataFrame(pd.DataFrame(data))
df.show()
agg_df = df.groupBy("incoming_ips").agg(F.sum("gbps").alias("total_gbps"))
windowRank = Window.orderBy(F.col("total_gbps").desc())
agg_df = agg_df.withColumn("rank", F.percent_rank().over(windowRank))
top_80_ips = agg_df.filter(F.col("rank") <= 0.80)
result = df.join(top_80_ips, on="incoming_ips", how="inner").select("incoming_ips", "gbps", "date_time")
result.show()
print(df.count())
print(result_df.count())
+---------------+-------------------+-------------------+
| incoming_ips| gbps| date_time|
+---------------+-------------------+-------------------+
| 66.186.8.130| 5.074283124722104|2022-03-12 05:09:16|
| 155.45.76.235| 0.6736194760917324|2021-06-19 03:36:28|
| 237.51.137.200|0.43334812775057685|2022-04-27 08:08:47|
| 78.4.48.171| 7.5675453578753435|2022-08-21 18:55:48|
| 241.84.163.17| 3.5681655964070815|2021-01-24 20:39:50|
|130.255.202.138| 6.066112278135983|2023-07-07 22:26:15|
| 198.33.206.140| 1.9147905257021836|2023-03-01 04:44:14|
| 84.183.253.20| 7.707176860385722|2021-08-26 23:24:31|
|218.163.165.232| 9.458673015973213|2021-02-22 12:13:15|
| 62.57.20.153| 1.5764916247359229|2021-11-06 12:41:59|
| 245.24.168.152|0.07452805411698016|2021-06-04 16:14:36|
| 98.171.202.249| 3.546118349483626|2022-07-05 10:55:26|
| 210.5.246.85|0.02430730260109759|2022-04-08 17:26:04|
| 13.236.170.177| 2.41361938344535|2021-08-11 02:19:06|
|180.140.248.193| 0.9512956363005021|2021-06-27 18:16:58|
| 26.140.88.127| 7.51335778127692|2023-06-02 14:13:30|
| 7.118.207.252| 6.450499049816286|2022-12-11 06:36:20|
| 11.8.10.136| 8.750329246667354|2023-02-03 05:33:16|
| 232.140.56.86| 4.289740988237201|2023-02-22 20:10:09|
| 68.117.9.255| 5.384340363304169|2022-12-03 09:55:26|
+---------------+-------------------+-------------------+
+---------------+------------------+-------------------+
| incoming_ips| gbps| date_time|
+---------------+------------------+-------------------+
| 66.186.8.130| 5.074283124722104|2022-03-12 05:09:16|
| 241.84.163.17|3.5681655964070815|2021-01-24 20:39:50|
| 78.4.48.171|7.5675453578753435|2022-08-21 18:55:48|
|130.255.202.138| 6.066112278135983|2023-07-07 22:26:15|
| 198.33.206.140|1.9147905257021836|2023-03-01 04:44:14|
| 84.183.253.20| 7.707176860385722|2021-08-26 23:24:31|
|218.163.165.232| 9.458673015973213|2021-02-22 12:13:15|
| 62.57.20.153|1.5764916247359229|2021-11-06 12:41:59|
| 98.171.202.249| 3.546118349483626|2022-07-05 10:55:26|
|180.140.248.193|0.9512956363005021|2021-06-27 18:16:58|
| 13.236.170.177| 2.41361938344535|2021-08-11 02:19:06|
| 26.140.88.127| 7.51335778127692|2023-06-02 14:13:30|
| 7.118.207.252| 6.450499049816286|2022-12-11 06:36:20|
| 11.8.10.136| 8.750329246667354|2023-02-03 05:33:16|
| 232.140.56.86| 4.289740988237201|2023-02-22 20:10:09|
| 68.117.9.255| 5.384340363304169|2022-12-03 09:55:26|
+---------------+------------------+-------------------+
20
16
fre. 15. sep. 2023 kl. 20:14 skrev ashok34668@yahoo.com.INVALID <as...@yahoo.com.invalid>:
Hi team,
I am using PySpark 3.4
I have a table of million rows that has few columns. among them incoming ips and what is known as gbps (Gigabytes per second) and date and time of incoming ip.
I want to filter out 20% of low active ips and work on the remainder of data. How can I do thiis in PySpark?
Thanks
--
Bjørn Jørgensen
Vestre Aspehaug 4, 6010 Ålesund
Norge
+47 480 94 297
--
Bjørn Jørgensen
Vestre Aspehaug 4, 6010 Ålesund
Norge
+47 480 94 297
Re: Filter out 20% of rows
Posted by Mich Talebzadeh <mi...@gmail.com>.
Hi Bjorn,
I thought that one is better off using percentile_approx as it seems to be
the recommended approach for computing percentiles and can simplify the
code.
I have modified your code to use percentile_approx rather than manually
computing it. It would be interesting to hear ideas on this.
Here is the code:
# Standard library imports
import json
import multiprocessing
import os
import re
import sys
import random
# Third-party imports
import numpy as np
import pandas as pd
import pyarrow
# Pyspark imports
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession, functions as F, Window
from pyspark.sql.functions import (
col, concat, concat_ws, expr, lit, trim, udf
)
from pyspark.sql.types import (
IntegerType, StringType, StructField, StructType,
DoubleType, TimestampType
)
from pyspark import pandas as ps
os.environ["PYARROW_IGNORE_TIMEZONE"] = "1"
number_cores = int(multiprocessing.cpu_count())
mem_bytes = os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES") #
e.g. 4015976448
memory_gb = int(mem_bytes / (1024.0**3)) # e.g. 3.74
def get_spark_session(app_name: str, conf: SparkConf):
conf.setMaster("local[{}]".format(number_cores))
conf.set("spark.driver.memory", "{}g".format(memory_gb)).set(
"spark.sql.repl.eagerEval.enabled", "True"
).set("spark.sql.adaptive.enabled", "True").set(
"spark.serializer", "org.apache.spark.serializer.KryoSerializer"
).set(
"spark.sql.repl.eagerEval.maxNumRows", "10000"
)
return
SparkSession.builder.appName(app_name).config(conf=conf).getOrCreate()
spark = get_spark_session("My super app", SparkConf())
sc = SparkContext.getOrCreate()
sc.setLogLevel("ERROR")
def generate_ip():
return ".".join(str(random.randint(0, 255)) for _ in range(4))
def generate_timestamp():
return pd.Timestamp(
year=random.randint(2021, 2023),
month=random.randint(1, 12),
day=random.randint(1, 28),
hour=random.randint(0, 23),
minute=random.randint(0, 59),
second=random.randint(0, 59)
)
def random_gbps():
return random.uniform(0, 10)
# Number of rows
n = 20
data = [{"incoming_ips": generate_ip(), "gbps": random_gbps(), "date_time":
generate_timestamp()} for _ in range(n)]
df = spark.createDataFrame(pd.DataFrame(data))
df.show()
agg_df = df.groupBy("incoming_ips").agg(F.sum("gbps").alias("total_gbps"))
windowRank = Window.orderBy(F.col("total_gbps").desc())
agg_df = agg_df.withColumn("rank", F.percent_rank().over(windowRank))
# Calculate the 80th percentile value
percentile_80 = agg_df.agg(F.expr("percentile_approx(total_gbps,
0.8)").alias("percentile_80")).collect()[0]["percentile_80"]
# Filter the DataFrame based on the condition
filtered_df = df.filter(df["gbps"] >= percentile_80)
# Show the filtered DataFrame
print(f"Filtered DataFrame")
filtered_df.show()
print(f"Total rows in data frame = {df.count()}")
print(f"Result satisfying 80% percentile = {filtered_df.count()}")
And this is the results
+---------------+------------------+-------------------+
| incoming_ips| gbps| date_time|
+---------------+------------------+-------------------+
|129.189.130.141|2.6517421918102335|2021-09-06 08:29:25|
| 215.177.39.239|1.8210013026429361|2023-10-10 17:00:13|
| 78.202.71.184| 8.060958370556456|2022-02-22 04:25:03|
|219.100.198.137|0.3449002002472945|2023-09-28 01:39:44|
|234.234.156.107|2.6187481766507013|2022-11-16 11:33:41|
| 6.223.135.194|0.3510752223686242|2022-01-24 04:13:53|
| 147.118.171.59|6.4071750880652765|2023-10-08 16:49:10|
| 75.41.101.165|2.1484984272041685|2022-07-13 21:02:58|
| 163.26.238.22| 9.8999646499433|2023-01-12 17:54:44|
| 184.145.98.231|1.8875849709728088|2022-09-18 19:53:58|
| 125.77.236.177| 1.17126350326476|2021-08-19 18:48:42|
| 34.103.211.39| 9.51081430594299|2023-02-05 18:39:23|
| 117.37.42.91| 1.122437784309721|2021-03-23 17:27:27|
| 108.115.42.171| 8.165187506266607|2023-07-26 03:57:50|
| 98.105.153.129| 9.284242190156004|2023-10-10 22:36:47|
| 145.35.252.142| 9.787384042283957|2022-08-26 00:53:27|
| 18.76.138.108| 6.939770760444909|2022-04-01 01:18:27|
| 31.33.71.26| 4.820947188427366|2021-06-10 22:02:51|
| 135.22.8.38| 9.587849542001745|2021-09-21 15:11:59|
|104.231.110.207| 9.045897927807715|2023-06-28 06:01:00|
+---------------+------------------+-------------------+
Filtered DataFrame
+--------------+-----------------+-------------------+
| incoming_ips| gbps| date_time|
+--------------+-----------------+-------------------+
| 163.26.238.22| 9.8999646499433|2023-01-12 17:54:44|
| 34.103.211.39| 9.51081430594299|2023-02-05 18:39:23|
|98.105.153.129|9.284242190156004|2023-10-10 22:36:47|
|145.35.252.142|9.787384042283957|2022-08-26 00:53:27|
| 135.22.8.38|9.587849542001745|2021-09-21 15:11:59|
+--------------+-----------------+-------------------+
Total rows in data frame = 20
Result satisfying 80% percentile = 5
Cheers
Mich Talebzadeh,
Distinguished Technologist, Solutions Architect & Engineer
London
United Kingdom
view my Linkedin profile
<https://www.linkedin.com/in/mich-talebzadeh-ph-d-5205b2/>
https://en.everybodywiki.com/Mich_Talebzadeh
*Disclaimer:* Use it at your own risk. Any and all responsibility for any
loss, damage or destruction of data or any other property which may arise
from relying on this email's technical content is explicitly disclaimed.
The author will in no case be liable for any monetary damages arising from
such loss, damage or destruction.
On Sat, 16 Sept 2023 at 11:46, Mich Talebzadeh <mi...@gmail.com>
wrote:
> Happy Saturday coding 😁
>
>
> Mich Talebzadeh,
> Distinguished Technologist, Solutions Architect & Engineer
> London
> United Kingdom
>
>
> view my Linkedin profile
> <https://www.linkedin.com/in/mich-talebzadeh-ph-d-5205b2/>
>
>
> https://en.everybodywiki.com/Mich_Talebzadeh
>
>
>
> *Disclaimer:* Use it at your own risk. Any and all responsibility for any
> loss, damage or destruction of data or any other property which may arise
> from relying on this email's technical content is explicitly disclaimed.
> The author will in no case be liable for any monetary damages arising from
> such loss, damage or destruction.
>
>
>
>
> On Sat, 16 Sept 2023 at 11:30, Bjørn Jørgensen <bj...@gmail.com>
> wrote:
>
>> ah.. yes that's right.
>> I did have to use some time on this one and I was having some issues with
>> the code.
>> I restart the notebook kernel now and rerun it and I get the same
>> result.
>>
>> lør. 16. sep. 2023 kl. 11:41 skrev Mich Talebzadeh <
>> mich.talebzadeh@gmail.com>:
>>
>>> Splendid code. A minor error glancing at your code.
>>>
>>> print(df.count())
>>> print(result_df.count())
>>>
>>>
>>> You have not defined result_df. I gather you meant "result"?
>>>
>>>
>>> print(result.count())
>>>
>>>
>>> That should fix it 🤔
>>>
>>> HTH
>>>
>>>
>>> Mich Talebzadeh,
>>> Distinguished Technologist, Solutions Architect & Engineer
>>> London
>>> United Kingdom
>>>
>>>
>>> view my Linkedin profile
>>> <https://www.linkedin.com/in/mich-talebzadeh-ph-d-5205b2/>
>>>
>>>
>>> https://en.everybodywiki.com/Mich_Talebzadeh
>>>
>>>
>>>
>>> *Disclaimer:* Use it at your own risk. Any and all responsibility for
>>> any loss, damage or destruction of data or any other property which may
>>> arise from relying on this email's technical content is explicitly
>>> disclaimed. The author will in no case be liable for any monetary damages
>>> arising from such loss, damage or destruction.
>>>
>>>
>>>
>>>
>>> On Sat, 16 Sept 2023 at 06:00, Bjørn Jørgensen <bj...@gmail.com>
>>> wrote:
>>>
>>>> Something like this?
>>>>
>>>>
>>>> # Standard library imports
>>>> import json
>>>> import multiprocessing
>>>> import os
>>>> import re
>>>> import sys
>>>> import random
>>>>
>>>> # Third-party imports
>>>> import numpy as np
>>>> import pandas as pd
>>>> import pyarrow
>>>>
>>>> # Pyspark imports
>>>> from pyspark import SparkConf, SparkContext
>>>> from pyspark.sql import SparkSession, functions as F, Window
>>>> from pyspark.sql.functions import (
>>>> col, concat, concat_ws, expr, lit, trim, udf
>>>> )
>>>> from pyspark.sql.types import (
>>>> IntegerType, StringType, StructField, StructType,
>>>> DoubleType, TimestampType
>>>> )
>>>> from pyspark import pandas as ps
>>>>
>>>> os.environ["PYARROW_IGNORE_TIMEZONE"] = "1"
>>>>
>>>> number_cores = int(multiprocessing.cpu_count())
>>>>
>>>> mem_bytes = os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES") # e.g. 4015976448
>>>> memory_gb = int(mem_bytes / (1024.0**3)) # e.g. 3.74
>>>>
>>>>
>>>> def get_spark_session(app_name: str, conf: SparkConf):
>>>> conf.setMaster("local[{}]".format(number_cores))
>>>> conf.set("spark.driver.memory", "{}g".format(memory_gb)).set(
>>>> "spark.sql.repl.eagerEval.enabled", "True"
>>>> ).set("spark.sql.adaptive.enabled", "True").set(
>>>> "spark.serializer", "org.apache.spark.serializer.KryoSerializer"
>>>> ).set(
>>>> "spark.sql.repl.eagerEval.maxNumRows", "10000"
>>>> ).set(
>>>> "sc.setLogLevel", "ERROR"
>>>> )
>>>>
>>>> return SparkSession.builder.appName(app_name).config(conf=conf).getOrCreate()
>>>>
>>>>
>>>> spark = get_spark_session("My super app", SparkConf())
>>>> spark.sparkContext.setLogLevel("ERROR")
>>>>
>>>>
>>>>
>>>> def generate_ip():
>>>> return ".".join(str(random.randint(0, 255)) for _ in range(4))
>>>>
>>>> def generate_timestamp():
>>>> return pd.Timestamp(
>>>> year=random.randint(2021, 2023),
>>>> month=random.randint(1, 12),
>>>> day=random.randint(1, 28),
>>>> hour=random.randint(0, 23),
>>>> minute=random.randint(0, 59),
>>>> second=random.randint(0, 59)
>>>> )
>>>>
>>>> def random_gbps():
>>>> return random.uniform(0, 10)
>>>>
>>>> # Number of rows
>>>> n = 20
>>>>
>>>> data = [{"incoming_ips": generate_ip(), "gbps": random_gbps(), "date_time": generate_timestamp()} for _ in range(n)]
>>>> df = spark.createDataFrame(pd.DataFrame(data))
>>>> df.show()
>>>>
>>>> agg_df = df.groupBy("incoming_ips").agg(F.sum("gbps").alias("total_gbps"))
>>>>
>>>> windowRank = Window.orderBy(F.col("total_gbps").desc())
>>>> agg_df = agg_df.withColumn("rank", F.percent_rank().over(windowRank))
>>>>
>>>> top_80_ips = agg_df.filter(F.col("rank") <= 0.80)
>>>> result = df.join(top_80_ips, on="incoming_ips", how="inner").select("incoming_ips", "gbps", "date_time")
>>>> result.show()
>>>>
>>>> print(df.count())
>>>> print(result_df.count())
>>>>
>>>>
>>>> +---------------+-------------------+-------------------+
>>>> | incoming_ips| gbps| date_time|
>>>> +---------------+-------------------+-------------------+
>>>> | 66.186.8.130| 5.074283124722104|2022-03-12 05:09:16|
>>>> | 155.45.76.235| 0.6736194760917324|2021-06-19 03:36:28|
>>>> | 237.51.137.200|0.43334812775057685|2022-04-27 08:08:47|
>>>> | 78.4.48.171| 7.5675453578753435|2022-08-21 18:55:48|
>>>> | 241.84.163.17| 3.5681655964070815|2021-01-24 20:39:50|
>>>> |130.255.202.138| 6.066112278135983|2023-07-07 22:26:15|
>>>> | 198.33.206.140| 1.9147905257021836|2023-03-01 04:44:14|
>>>> | 84.183.253.20| 7.707176860385722|2021-08-26 23:24:31|
>>>> |218.163.165.232| 9.458673015973213|2021-02-22 12:13:15|
>>>> | 62.57.20.153| 1.5764916247359229|2021-11-06 12:41:59|
>>>> | 245.24.168.152|0.07452805411698016|2021-06-04 16:14:36|
>>>> | 98.171.202.249| 3.546118349483626|2022-07-05 10:55:26|
>>>> | 210.5.246.85|0.02430730260109759|2022-04-08 17:26:04|
>>>> | 13.236.170.177| 2.41361938344535|2021-08-11 02:19:06|
>>>> |180.140.248.193| 0.9512956363005021|2021-06-27 18:16:58|
>>>> | 26.140.88.127| 7.51335778127692|2023-06-02 14:13:30|
>>>> | 7.118.207.252| 6.450499049816286|2022-12-11 06:36:20|
>>>> | 11.8.10.136| 8.750329246667354|2023-02-03 05:33:16|
>>>> | 232.140.56.86| 4.289740988237201|2023-02-22 20:10:09|
>>>> | 68.117.9.255| 5.384340363304169|2022-12-03 09:55:26|
>>>> +---------------+-------------------+-------------------+
>>>>
>>>> +---------------+------------------+-------------------+
>>>> | incoming_ips| gbps| date_time|
>>>> +---------------+------------------+-------------------+
>>>> | 66.186.8.130| 5.074283124722104|2022-03-12 05:09:16|
>>>> | 241.84.163.17|3.5681655964070815|2021-01-24 20:39:50|
>>>> | 78.4.48.171|7.5675453578753435|2022-08-21 18:55:48|
>>>> |130.255.202.138| 6.066112278135983|2023-07-07 22:26:15|
>>>> | 198.33.206.140|1.9147905257021836|2023-03-01 04:44:14|
>>>> | 84.183.253.20| 7.707176860385722|2021-08-26 23:24:31|
>>>> |218.163.165.232| 9.458673015973213|2021-02-22 12:13:15|
>>>> | 62.57.20.153|1.5764916247359229|2021-11-06 12:41:59|
>>>> | 98.171.202.249| 3.546118349483626|2022-07-05 10:55:26|
>>>> |180.140.248.193|0.9512956363005021|2021-06-27 18:16:58|
>>>> | 13.236.170.177| 2.41361938344535|2021-08-11 02:19:06|
>>>> | 26.140.88.127| 7.51335778127692|2023-06-02 14:13:30|
>>>> | 7.118.207.252| 6.450499049816286|2022-12-11 06:36:20|
>>>> | 11.8.10.136| 8.750329246667354|2023-02-03 05:33:16|
>>>> | 232.140.56.86| 4.289740988237201|2023-02-22 20:10:09|
>>>> | 68.117.9.255| 5.384340363304169|2022-12-03 09:55:26|
>>>> +---------------+------------------+-------------------+
>>>>
>>>> 20
>>>> 16
>>>>
>>>>
>>>>
>>>> fre. 15. sep. 2023 kl. 20:14 skrev ashok34668@yahoo.com.INVALID
>>>> <as...@yahoo.com.invalid>:
>>>>
>>>>> Hi team,
>>>>>
>>>>> I am using PySpark 3.4
>>>>>
>>>>> I have a table of million rows that has few columns. among them
>>>>> incoming ips and what is known as gbps (Gigabytes per second) and
>>>>> date and time of incoming ip.
>>>>>
>>>>> I want to filter out 20% of low active ips and work on the remainder
>>>>> of data. How can I do thiis in PySpark?
>>>>>
>>>>> Thanks
>>>>>
>>>>>
>>>>>
>>>>>
>>>>>
>>>>
>>>> --
>>>> Bjørn Jørgensen
>>>> Vestre Aspehaug 4, 6010 Ålesund
>>>> Norge
>>>>
>>>> +47 480 94 297
>>>>
>>>
>>
>> --
>> Bjørn Jørgensen
>> Vestre Aspehaug 4, 6010 Ålesund
>> Norge
>>
>> +47 480 94 297
>>
>
Re: Filter out 20% of rows
Posted by Mich Talebzadeh <mi...@gmail.com>.
Happy Saturday coding 😁
Mich Talebzadeh,
Distinguished Technologist, Solutions Architect & Engineer
London
United Kingdom
view my Linkedin profile
<https://www.linkedin.com/in/mich-talebzadeh-ph-d-5205b2/>
https://en.everybodywiki.com/Mich_Talebzadeh
*Disclaimer:* Use it at your own risk. Any and all responsibility for any
loss, damage or destruction of data or any other property which may arise
from relying on this email's technical content is explicitly disclaimed.
The author will in no case be liable for any monetary damages arising from
such loss, damage or destruction.
On Sat, 16 Sept 2023 at 11:30, Bjørn Jørgensen <bj...@gmail.com>
wrote:
> ah.. yes that's right.
> I did have to use some time on this one and I was having some issues with
> the code.
> I restart the notebook kernel now and rerun it and I get the same result.
>
> lør. 16. sep. 2023 kl. 11:41 skrev Mich Talebzadeh <
> mich.talebzadeh@gmail.com>:
>
>> Splendid code. A minor error glancing at your code.
>>
>> print(df.count())
>> print(result_df.count())
>>
>>
>> You have not defined result_df. I gather you meant "result"?
>>
>>
>> print(result.count())
>>
>>
>> That should fix it 🤔
>>
>> HTH
>>
>>
>> Mich Talebzadeh,
>> Distinguished Technologist, Solutions Architect & Engineer
>> London
>> United Kingdom
>>
>>
>> view my Linkedin profile
>> <https://www.linkedin.com/in/mich-talebzadeh-ph-d-5205b2/>
>>
>>
>> https://en.everybodywiki.com/Mich_Talebzadeh
>>
>>
>>
>> *Disclaimer:* Use it at your own risk. Any and all responsibility for
>> any loss, damage or destruction of data or any other property which may
>> arise from relying on this email's technical content is explicitly
>> disclaimed. The author will in no case be liable for any monetary damages
>> arising from such loss, damage or destruction.
>>
>>
>>
>>
>> On Sat, 16 Sept 2023 at 06:00, Bjørn Jørgensen <bj...@gmail.com>
>> wrote:
>>
>>> Something like this?
>>>
>>>
>>> # Standard library imports
>>> import json
>>> import multiprocessing
>>> import os
>>> import re
>>> import sys
>>> import random
>>>
>>> # Third-party imports
>>> import numpy as np
>>> import pandas as pd
>>> import pyarrow
>>>
>>> # Pyspark imports
>>> from pyspark import SparkConf, SparkContext
>>> from pyspark.sql import SparkSession, functions as F, Window
>>> from pyspark.sql.functions import (
>>> col, concat, concat_ws, expr, lit, trim, udf
>>> )
>>> from pyspark.sql.types import (
>>> IntegerType, StringType, StructField, StructType,
>>> DoubleType, TimestampType
>>> )
>>> from pyspark import pandas as ps
>>>
>>> os.environ["PYARROW_IGNORE_TIMEZONE"] = "1"
>>>
>>> number_cores = int(multiprocessing.cpu_count())
>>>
>>> mem_bytes = os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES") # e.g. 4015976448
>>> memory_gb = int(mem_bytes / (1024.0**3)) # e.g. 3.74
>>>
>>>
>>> def get_spark_session(app_name: str, conf: SparkConf):
>>> conf.setMaster("local[{}]".format(number_cores))
>>> conf.set("spark.driver.memory", "{}g".format(memory_gb)).set(
>>> "spark.sql.repl.eagerEval.enabled", "True"
>>> ).set("spark.sql.adaptive.enabled", "True").set(
>>> "spark.serializer", "org.apache.spark.serializer.KryoSerializer"
>>> ).set(
>>> "spark.sql.repl.eagerEval.maxNumRows", "10000"
>>> ).set(
>>> "sc.setLogLevel", "ERROR"
>>> )
>>>
>>> return SparkSession.builder.appName(app_name).config(conf=conf).getOrCreate()
>>>
>>>
>>> spark = get_spark_session("My super app", SparkConf())
>>> spark.sparkContext.setLogLevel("ERROR")
>>>
>>>
>>>
>>> def generate_ip():
>>> return ".".join(str(random.randint(0, 255)) for _ in range(4))
>>>
>>> def generate_timestamp():
>>> return pd.Timestamp(
>>> year=random.randint(2021, 2023),
>>> month=random.randint(1, 12),
>>> day=random.randint(1, 28),
>>> hour=random.randint(0, 23),
>>> minute=random.randint(0, 59),
>>> second=random.randint(0, 59)
>>> )
>>>
>>> def random_gbps():
>>> return random.uniform(0, 10)
>>>
>>> # Number of rows
>>> n = 20
>>>
>>> data = [{"incoming_ips": generate_ip(), "gbps": random_gbps(), "date_time": generate_timestamp()} for _ in range(n)]
>>> df = spark.createDataFrame(pd.DataFrame(data))
>>> df.show()
>>>
>>> agg_df = df.groupBy("incoming_ips").agg(F.sum("gbps").alias("total_gbps"))
>>>
>>> windowRank = Window.orderBy(F.col("total_gbps").desc())
>>> agg_df = agg_df.withColumn("rank", F.percent_rank().over(windowRank))
>>>
>>> top_80_ips = agg_df.filter(F.col("rank") <= 0.80)
>>> result = df.join(top_80_ips, on="incoming_ips", how="inner").select("incoming_ips", "gbps", "date_time")
>>> result.show()
>>>
>>> print(df.count())
>>> print(result_df.count())
>>>
>>>
>>> +---------------+-------------------+-------------------+
>>> | incoming_ips| gbps| date_time|
>>> +---------------+-------------------+-------------------+
>>> | 66.186.8.130| 5.074283124722104|2022-03-12 05:09:16|
>>> | 155.45.76.235| 0.6736194760917324|2021-06-19 03:36:28|
>>> | 237.51.137.200|0.43334812775057685|2022-04-27 08:08:47|
>>> | 78.4.48.171| 7.5675453578753435|2022-08-21 18:55:48|
>>> | 241.84.163.17| 3.5681655964070815|2021-01-24 20:39:50|
>>> |130.255.202.138| 6.066112278135983|2023-07-07 22:26:15|
>>> | 198.33.206.140| 1.9147905257021836|2023-03-01 04:44:14|
>>> | 84.183.253.20| 7.707176860385722|2021-08-26 23:24:31|
>>> |218.163.165.232| 9.458673015973213|2021-02-22 12:13:15|
>>> | 62.57.20.153| 1.5764916247359229|2021-11-06 12:41:59|
>>> | 245.24.168.152|0.07452805411698016|2021-06-04 16:14:36|
>>> | 98.171.202.249| 3.546118349483626|2022-07-05 10:55:26|
>>> | 210.5.246.85|0.02430730260109759|2022-04-08 17:26:04|
>>> | 13.236.170.177| 2.41361938344535|2021-08-11 02:19:06|
>>> |180.140.248.193| 0.9512956363005021|2021-06-27 18:16:58|
>>> | 26.140.88.127| 7.51335778127692|2023-06-02 14:13:30|
>>> | 7.118.207.252| 6.450499049816286|2022-12-11 06:36:20|
>>> | 11.8.10.136| 8.750329246667354|2023-02-03 05:33:16|
>>> | 232.140.56.86| 4.289740988237201|2023-02-22 20:10:09|
>>> | 68.117.9.255| 5.384340363304169|2022-12-03 09:55:26|
>>> +---------------+-------------------+-------------------+
>>>
>>> +---------------+------------------+-------------------+
>>> | incoming_ips| gbps| date_time|
>>> +---------------+------------------+-------------------+
>>> | 66.186.8.130| 5.074283124722104|2022-03-12 05:09:16|
>>> | 241.84.163.17|3.5681655964070815|2021-01-24 20:39:50|
>>> | 78.4.48.171|7.5675453578753435|2022-08-21 18:55:48|
>>> |130.255.202.138| 6.066112278135983|2023-07-07 22:26:15|
>>> | 198.33.206.140|1.9147905257021836|2023-03-01 04:44:14|
>>> | 84.183.253.20| 7.707176860385722|2021-08-26 23:24:31|
>>> |218.163.165.232| 9.458673015973213|2021-02-22 12:13:15|
>>> | 62.57.20.153|1.5764916247359229|2021-11-06 12:41:59|
>>> | 98.171.202.249| 3.546118349483626|2022-07-05 10:55:26|
>>> |180.140.248.193|0.9512956363005021|2021-06-27 18:16:58|
>>> | 13.236.170.177| 2.41361938344535|2021-08-11 02:19:06|
>>> | 26.140.88.127| 7.51335778127692|2023-06-02 14:13:30|
>>> | 7.118.207.252| 6.450499049816286|2022-12-11 06:36:20|
>>> | 11.8.10.136| 8.750329246667354|2023-02-03 05:33:16|
>>> | 232.140.56.86| 4.289740988237201|2023-02-22 20:10:09|
>>> | 68.117.9.255| 5.384340363304169|2022-12-03 09:55:26|
>>> +---------------+------------------+-------------------+
>>>
>>> 20
>>> 16
>>>
>>>
>>>
>>> fre. 15. sep. 2023 kl. 20:14 skrev ashok34668@yahoo.com.INVALID
>>> <as...@yahoo.com.invalid>:
>>>
>>>> Hi team,
>>>>
>>>> I am using PySpark 3.4
>>>>
>>>> I have a table of million rows that has few columns. among them
>>>> incoming ips and what is known as gbps (Gigabytes per second) and
>>>> date and time of incoming ip.
>>>>
>>>> I want to filter out 20% of low active ips and work on the remainder of
>>>> data. How can I do thiis in PySpark?
>>>>
>>>> Thanks
>>>>
>>>>
>>>>
>>>>
>>>>
>>>
>>> --
>>> Bjørn Jørgensen
>>> Vestre Aspehaug 4, 6010 Ålesund
>>> Norge
>>>
>>> +47 480 94 297
>>>
>>
>
> --
> Bjørn Jørgensen
> Vestre Aspehaug 4, 6010 Ålesund
> Norge
>
> +47 480 94 297
>
Re: Filter out 20% of rows
Posted by Bjørn Jørgensen <bj...@gmail.com>.
ah.. yes that's right.
I did have to use some time on this one and I was having some issues with
the code.
I restart the notebook kernel now and rerun it and I get the same result.
lør. 16. sep. 2023 kl. 11:41 skrev Mich Talebzadeh <
mich.talebzadeh@gmail.com>:
> Splendid code. A minor error glancing at your code.
>
> print(df.count())
> print(result_df.count())
>
>
> You have not defined result_df. I gather you meant "result"?
>
>
> print(result.count())
>
>
> That should fix it 🤔
>
> HTH
>
>
> Mich Talebzadeh,
> Distinguished Technologist, Solutions Architect & Engineer
> London
> United Kingdom
>
>
> view my Linkedin profile
> <https://www.linkedin.com/in/mich-talebzadeh-ph-d-5205b2/>
>
>
> https://en.everybodywiki.com/Mich_Talebzadeh
>
>
>
> *Disclaimer:* Use it at your own risk. Any and all responsibility for any
> loss, damage or destruction of data or any other property which may arise
> from relying on this email's technical content is explicitly disclaimed.
> The author will in no case be liable for any monetary damages arising from
> such loss, damage or destruction.
>
>
>
>
> On Sat, 16 Sept 2023 at 06:00, Bjørn Jørgensen <bj...@gmail.com>
> wrote:
>
>> Something like this?
>>
>>
>> # Standard library imports
>> import json
>> import multiprocessing
>> import os
>> import re
>> import sys
>> import random
>>
>> # Third-party imports
>> import numpy as np
>> import pandas as pd
>> import pyarrow
>>
>> # Pyspark imports
>> from pyspark import SparkConf, SparkContext
>> from pyspark.sql import SparkSession, functions as F, Window
>> from pyspark.sql.functions import (
>> col, concat, concat_ws, expr, lit, trim, udf
>> )
>> from pyspark.sql.types import (
>> IntegerType, StringType, StructField, StructType,
>> DoubleType, TimestampType
>> )
>> from pyspark import pandas as ps
>>
>> os.environ["PYARROW_IGNORE_TIMEZONE"] = "1"
>>
>> number_cores = int(multiprocessing.cpu_count())
>>
>> mem_bytes = os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES") # e.g. 4015976448
>> memory_gb = int(mem_bytes / (1024.0**3)) # e.g. 3.74
>>
>>
>> def get_spark_session(app_name: str, conf: SparkConf):
>> conf.setMaster("local[{}]".format(number_cores))
>> conf.set("spark.driver.memory", "{}g".format(memory_gb)).set(
>> "spark.sql.repl.eagerEval.enabled", "True"
>> ).set("spark.sql.adaptive.enabled", "True").set(
>> "spark.serializer", "org.apache.spark.serializer.KryoSerializer"
>> ).set(
>> "spark.sql.repl.eagerEval.maxNumRows", "10000"
>> ).set(
>> "sc.setLogLevel", "ERROR"
>> )
>>
>> return SparkSession.builder.appName(app_name).config(conf=conf).getOrCreate()
>>
>>
>> spark = get_spark_session("My super app", SparkConf())
>> spark.sparkContext.setLogLevel("ERROR")
>>
>>
>>
>> def generate_ip():
>> return ".".join(str(random.randint(0, 255)) for _ in range(4))
>>
>> def generate_timestamp():
>> return pd.Timestamp(
>> year=random.randint(2021, 2023),
>> month=random.randint(1, 12),
>> day=random.randint(1, 28),
>> hour=random.randint(0, 23),
>> minute=random.randint(0, 59),
>> second=random.randint(0, 59)
>> )
>>
>> def random_gbps():
>> return random.uniform(0, 10)
>>
>> # Number of rows
>> n = 20
>>
>> data = [{"incoming_ips": generate_ip(), "gbps": random_gbps(), "date_time": generate_timestamp()} for _ in range(n)]
>> df = spark.createDataFrame(pd.DataFrame(data))
>> df.show()
>>
>> agg_df = df.groupBy("incoming_ips").agg(F.sum("gbps").alias("total_gbps"))
>>
>> windowRank = Window.orderBy(F.col("total_gbps").desc())
>> agg_df = agg_df.withColumn("rank", F.percent_rank().over(windowRank))
>>
>> top_80_ips = agg_df.filter(F.col("rank") <= 0.80)
>> result = df.join(top_80_ips, on="incoming_ips", how="inner").select("incoming_ips", "gbps", "date_time")
>> result.show()
>>
>> print(df.count())
>> print(result_df.count())
>>
>>
>> +---------------+-------------------+-------------------+
>> | incoming_ips| gbps| date_time|
>> +---------------+-------------------+-------------------+
>> | 66.186.8.130| 5.074283124722104|2022-03-12 05:09:16|
>> | 155.45.76.235| 0.6736194760917324|2021-06-19 03:36:28|
>> | 237.51.137.200|0.43334812775057685|2022-04-27 08:08:47|
>> | 78.4.48.171| 7.5675453578753435|2022-08-21 18:55:48|
>> | 241.84.163.17| 3.5681655964070815|2021-01-24 20:39:50|
>> |130.255.202.138| 6.066112278135983|2023-07-07 22:26:15|
>> | 198.33.206.140| 1.9147905257021836|2023-03-01 04:44:14|
>> | 84.183.253.20| 7.707176860385722|2021-08-26 23:24:31|
>> |218.163.165.232| 9.458673015973213|2021-02-22 12:13:15|
>> | 62.57.20.153| 1.5764916247359229|2021-11-06 12:41:59|
>> | 245.24.168.152|0.07452805411698016|2021-06-04 16:14:36|
>> | 98.171.202.249| 3.546118349483626|2022-07-05 10:55:26|
>> | 210.5.246.85|0.02430730260109759|2022-04-08 17:26:04|
>> | 13.236.170.177| 2.41361938344535|2021-08-11 02:19:06|
>> |180.140.248.193| 0.9512956363005021|2021-06-27 18:16:58|
>> | 26.140.88.127| 7.51335778127692|2023-06-02 14:13:30|
>> | 7.118.207.252| 6.450499049816286|2022-12-11 06:36:20|
>> | 11.8.10.136| 8.750329246667354|2023-02-03 05:33:16|
>> | 232.140.56.86| 4.289740988237201|2023-02-22 20:10:09|
>> | 68.117.9.255| 5.384340363304169|2022-12-03 09:55:26|
>> +---------------+-------------------+-------------------+
>>
>> +---------------+------------------+-------------------+
>> | incoming_ips| gbps| date_time|
>> +---------------+------------------+-------------------+
>> | 66.186.8.130| 5.074283124722104|2022-03-12 05:09:16|
>> | 241.84.163.17|3.5681655964070815|2021-01-24 20:39:50|
>> | 78.4.48.171|7.5675453578753435|2022-08-21 18:55:48|
>> |130.255.202.138| 6.066112278135983|2023-07-07 22:26:15|
>> | 198.33.206.140|1.9147905257021836|2023-03-01 04:44:14|
>> | 84.183.253.20| 7.707176860385722|2021-08-26 23:24:31|
>> |218.163.165.232| 9.458673015973213|2021-02-22 12:13:15|
>> | 62.57.20.153|1.5764916247359229|2021-11-06 12:41:59|
>> | 98.171.202.249| 3.546118349483626|2022-07-05 10:55:26|
>> |180.140.248.193|0.9512956363005021|2021-06-27 18:16:58|
>> | 13.236.170.177| 2.41361938344535|2021-08-11 02:19:06|
>> | 26.140.88.127| 7.51335778127692|2023-06-02 14:13:30|
>> | 7.118.207.252| 6.450499049816286|2022-12-11 06:36:20|
>> | 11.8.10.136| 8.750329246667354|2023-02-03 05:33:16|
>> | 232.140.56.86| 4.289740988237201|2023-02-22 20:10:09|
>> | 68.117.9.255| 5.384340363304169|2022-12-03 09:55:26|
>> +---------------+------------------+-------------------+
>>
>> 20
>> 16
>>
>>
>>
>> fre. 15. sep. 2023 kl. 20:14 skrev ashok34668@yahoo.com.INVALID
>> <as...@yahoo.com.invalid>:
>>
>>> Hi team,
>>>
>>> I am using PySpark 3.4
>>>
>>> I have a table of million rows that has few columns. among them incoming
>>> ips and what is known as gbps (Gigabytes per second) and date and time
>>> of incoming ip.
>>>
>>> I want to filter out 20% of low active ips and work on the remainder of
>>> data. How can I do thiis in PySpark?
>>>
>>> Thanks
>>>
>>>
>>>
>>>
>>>
>>
>> --
>> Bjørn Jørgensen
>> Vestre Aspehaug 4, 6010 Ålesund
>> Norge
>>
>> +47 480 94 297
>>
>
--
Bjørn Jørgensen
Vestre Aspehaug 4, 6010 Ålesund
Norge
+47 480 94 297
Re: Filter out 20% of rows
Posted by Mich Talebzadeh <mi...@gmail.com>.
Splendid code. A minor error glancing at your code.
print(df.count())
print(result_df.count())
You have not defined result_df. I gather you meant "result"?
print(result.count())
That should fix it 🤔
HTH
Mich Talebzadeh,
Distinguished Technologist, Solutions Architect & Engineer
London
United Kingdom
view my Linkedin profile
<https://www.linkedin.com/in/mich-talebzadeh-ph-d-5205b2/>
https://en.everybodywiki.com/Mich_Talebzadeh
*Disclaimer:* Use it at your own risk. Any and all responsibility for any
loss, damage or destruction of data or any other property which may arise
from relying on this email's technical content is explicitly disclaimed.
The author will in no case be liable for any monetary damages arising from
such loss, damage or destruction.
On Sat, 16 Sept 2023 at 06:00, Bjørn Jørgensen <bj...@gmail.com>
wrote:
> Something like this?
>
>
> # Standard library imports
> import json
> import multiprocessing
> import os
> import re
> import sys
> import random
>
> # Third-party imports
> import numpy as np
> import pandas as pd
> import pyarrow
>
> # Pyspark imports
> from pyspark import SparkConf, SparkContext
> from pyspark.sql import SparkSession, functions as F, Window
> from pyspark.sql.functions import (
> col, concat, concat_ws, expr, lit, trim, udf
> )
> from pyspark.sql.types import (
> IntegerType, StringType, StructField, StructType,
> DoubleType, TimestampType
> )
> from pyspark import pandas as ps
>
> os.environ["PYARROW_IGNORE_TIMEZONE"] = "1"
>
> number_cores = int(multiprocessing.cpu_count())
>
> mem_bytes = os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES") # e.g. 4015976448
> memory_gb = int(mem_bytes / (1024.0**3)) # e.g. 3.74
>
>
> def get_spark_session(app_name: str, conf: SparkConf):
> conf.setMaster("local[{}]".format(number_cores))
> conf.set("spark.driver.memory", "{}g".format(memory_gb)).set(
> "spark.sql.repl.eagerEval.enabled", "True"
> ).set("spark.sql.adaptive.enabled", "True").set(
> "spark.serializer", "org.apache.spark.serializer.KryoSerializer"
> ).set(
> "spark.sql.repl.eagerEval.maxNumRows", "10000"
> ).set(
> "sc.setLogLevel", "ERROR"
> )
>
> return SparkSession.builder.appName(app_name).config(conf=conf).getOrCreate()
>
>
> spark = get_spark_session("My super app", SparkConf())
> spark.sparkContext.setLogLevel("ERROR")
>
>
>
> def generate_ip():
> return ".".join(str(random.randint(0, 255)) for _ in range(4))
>
> def generate_timestamp():
> return pd.Timestamp(
> year=random.randint(2021, 2023),
> month=random.randint(1, 12),
> day=random.randint(1, 28),
> hour=random.randint(0, 23),
> minute=random.randint(0, 59),
> second=random.randint(0, 59)
> )
>
> def random_gbps():
> return random.uniform(0, 10)
>
> # Number of rows
> n = 20
>
> data = [{"incoming_ips": generate_ip(), "gbps": random_gbps(), "date_time": generate_timestamp()} for _ in range(n)]
> df = spark.createDataFrame(pd.DataFrame(data))
> df.show()
>
> agg_df = df.groupBy("incoming_ips").agg(F.sum("gbps").alias("total_gbps"))
>
> windowRank = Window.orderBy(F.col("total_gbps").desc())
> agg_df = agg_df.withColumn("rank", F.percent_rank().over(windowRank))
>
> top_80_ips = agg_df.filter(F.col("rank") <= 0.80)
> result = df.join(top_80_ips, on="incoming_ips", how="inner").select("incoming_ips", "gbps", "date_time")
> result.show()
>
> print(df.count())
> print(result_df.count())
>
>
> +---------------+-------------------+-------------------+
> | incoming_ips| gbps| date_time|
> +---------------+-------------------+-------------------+
> | 66.186.8.130| 5.074283124722104|2022-03-12 05:09:16|
> | 155.45.76.235| 0.6736194760917324|2021-06-19 03:36:28|
> | 237.51.137.200|0.43334812775057685|2022-04-27 08:08:47|
> | 78.4.48.171| 7.5675453578753435|2022-08-21 18:55:48|
> | 241.84.163.17| 3.5681655964070815|2021-01-24 20:39:50|
> |130.255.202.138| 6.066112278135983|2023-07-07 22:26:15|
> | 198.33.206.140| 1.9147905257021836|2023-03-01 04:44:14|
> | 84.183.253.20| 7.707176860385722|2021-08-26 23:24:31|
> |218.163.165.232| 9.458673015973213|2021-02-22 12:13:15|
> | 62.57.20.153| 1.5764916247359229|2021-11-06 12:41:59|
> | 245.24.168.152|0.07452805411698016|2021-06-04 16:14:36|
> | 98.171.202.249| 3.546118349483626|2022-07-05 10:55:26|
> | 210.5.246.85|0.02430730260109759|2022-04-08 17:26:04|
> | 13.236.170.177| 2.41361938344535|2021-08-11 02:19:06|
> |180.140.248.193| 0.9512956363005021|2021-06-27 18:16:58|
> | 26.140.88.127| 7.51335778127692|2023-06-02 14:13:30|
> | 7.118.207.252| 6.450499049816286|2022-12-11 06:36:20|
> | 11.8.10.136| 8.750329246667354|2023-02-03 05:33:16|
> | 232.140.56.86| 4.289740988237201|2023-02-22 20:10:09|
> | 68.117.9.255| 5.384340363304169|2022-12-03 09:55:26|
> +---------------+-------------------+-------------------+
>
> +---------------+------------------+-------------------+
> | incoming_ips| gbps| date_time|
> +---------------+------------------+-------------------+
> | 66.186.8.130| 5.074283124722104|2022-03-12 05:09:16|
> | 241.84.163.17|3.5681655964070815|2021-01-24 20:39:50|
> | 78.4.48.171|7.5675453578753435|2022-08-21 18:55:48|
> |130.255.202.138| 6.066112278135983|2023-07-07 22:26:15|
> | 198.33.206.140|1.9147905257021836|2023-03-01 04:44:14|
> | 84.183.253.20| 7.707176860385722|2021-08-26 23:24:31|
> |218.163.165.232| 9.458673015973213|2021-02-22 12:13:15|
> | 62.57.20.153|1.5764916247359229|2021-11-06 12:41:59|
> | 98.171.202.249| 3.546118349483626|2022-07-05 10:55:26|
> |180.140.248.193|0.9512956363005021|2021-06-27 18:16:58|
> | 13.236.170.177| 2.41361938344535|2021-08-11 02:19:06|
> | 26.140.88.127| 7.51335778127692|2023-06-02 14:13:30|
> | 7.118.207.252| 6.450499049816286|2022-12-11 06:36:20|
> | 11.8.10.136| 8.750329246667354|2023-02-03 05:33:16|
> | 232.140.56.86| 4.289740988237201|2023-02-22 20:10:09|
> | 68.117.9.255| 5.384340363304169|2022-12-03 09:55:26|
> +---------------+------------------+-------------------+
>
> 20
> 16
>
>
>
> fre. 15. sep. 2023 kl. 20:14 skrev ashok34668@yahoo.com.INVALID
> <as...@yahoo.com.invalid>:
>
>> Hi team,
>>
>> I am using PySpark 3.4
>>
>> I have a table of million rows that has few columns. among them incoming
>> ips and what is known as gbps (Gigabytes per second) and date and time
>> of incoming ip.
>>
>> I want to filter out 20% of low active ips and work on the remainder of
>> data. How can I do thiis in PySpark?
>>
>> Thanks
>>
>>
>>
>>
>>
>
> --
> Bjørn Jørgensen
> Vestre Aspehaug 4, 6010 Ålesund
> Norge
>
> +47 480 94 297
>
Re: Filter out 20% of rows
Posted by Bjørn Jørgensen <bj...@gmail.com>.
Something like this?
# Standard library imports
import json
import multiprocessing
import os
import re
import sys
import random
# Third-party imports
import numpy as np
import pandas as pd
import pyarrow
# Pyspark imports
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession, functions as F, Window
from pyspark.sql.functions import (
col, concat, concat_ws, expr, lit, trim, udf
)
from pyspark.sql.types import (
IntegerType, StringType, StructField, StructType,
DoubleType, TimestampType
)
from pyspark import pandas as ps
os.environ["PYARROW_IGNORE_TIMEZONE"] = "1"
number_cores = int(multiprocessing.cpu_count())
mem_bytes = os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES")
# e.g. 4015976448
memory_gb = int(mem_bytes / (1024.0**3)) # e.g. 3.74
def get_spark_session(app_name: str, conf: SparkConf):
conf.setMaster("local[{}]".format(number_cores))
conf.set("spark.driver.memory", "{}g".format(memory_gb)).set(
"spark.sql.repl.eagerEval.enabled", "True"
).set("spark.sql.adaptive.enabled", "True").set(
"spark.serializer", "org.apache.spark.serializer.KryoSerializer"
).set(
"spark.sql.repl.eagerEval.maxNumRows", "10000"
).set(
"sc.setLogLevel", "ERROR"
)
return SparkSession.builder.appName(app_name).config(conf=conf).getOrCreate()
spark = get_spark_session("My super app", SparkConf())
spark.sparkContext.setLogLevel("ERROR")
def generate_ip():
return ".".join(str(random.randint(0, 255)) for _ in range(4))
def generate_timestamp():
return pd.Timestamp(
year=random.randint(2021, 2023),
month=random.randint(1, 12),
day=random.randint(1, 28),
hour=random.randint(0, 23),
minute=random.randint(0, 59),
second=random.randint(0, 59)
)
def random_gbps():
return random.uniform(0, 10)
# Number of rows
n = 20
data = [{"incoming_ips": generate_ip(), "gbps": random_gbps(),
"date_time": generate_timestamp()} for _ in range(n)]
df = spark.createDataFrame(pd.DataFrame(data))
df.show()
agg_df = df.groupBy("incoming_ips").agg(F.sum("gbps").alias("total_gbps"))
windowRank = Window.orderBy(F.col("total_gbps").desc())
agg_df = agg_df.withColumn("rank", F.percent_rank().over(windowRank))
top_80_ips = agg_df.filter(F.col("rank") <= 0.80)
result = df.join(top_80_ips, on="incoming_ips",
how="inner").select("incoming_ips", "gbps", "date_time")
result.show()
print(df.count())
print(result_df.count())
+---------------+-------------------+-------------------+
| incoming_ips| gbps| date_time|
+---------------+-------------------+-------------------+
| 66.186.8.130| 5.074283124722104|2022-03-12 05:09:16|
| 155.45.76.235| 0.6736194760917324|2021-06-19 03:36:28|
| 237.51.137.200|0.43334812775057685|2022-04-27 08:08:47|
| 78.4.48.171| 7.5675453578753435|2022-08-21 18:55:48|
| 241.84.163.17| 3.5681655964070815|2021-01-24 20:39:50|
|130.255.202.138| 6.066112278135983|2023-07-07 22:26:15|
| 198.33.206.140| 1.9147905257021836|2023-03-01 04:44:14|
| 84.183.253.20| 7.707176860385722|2021-08-26 23:24:31|
|218.163.165.232| 9.458673015973213|2021-02-22 12:13:15|
| 62.57.20.153| 1.5764916247359229|2021-11-06 12:41:59|
| 245.24.168.152|0.07452805411698016|2021-06-04 16:14:36|
| 98.171.202.249| 3.546118349483626|2022-07-05 10:55:26|
| 210.5.246.85|0.02430730260109759|2022-04-08 17:26:04|
| 13.236.170.177| 2.41361938344535|2021-08-11 02:19:06|
|180.140.248.193| 0.9512956363005021|2021-06-27 18:16:58|
| 26.140.88.127| 7.51335778127692|2023-06-02 14:13:30|
| 7.118.207.252| 6.450499049816286|2022-12-11 06:36:20|
| 11.8.10.136| 8.750329246667354|2023-02-03 05:33:16|
| 232.140.56.86| 4.289740988237201|2023-02-22 20:10:09|
| 68.117.9.255| 5.384340363304169|2022-12-03 09:55:26|
+---------------+-------------------+-------------------+
+---------------+------------------+-------------------+
| incoming_ips| gbps| date_time|
+---------------+------------------+-------------------+
| 66.186.8.130| 5.074283124722104|2022-03-12 05:09:16|
| 241.84.163.17|3.5681655964070815|2021-01-24 20:39:50|
| 78.4.48.171|7.5675453578753435|2022-08-21 18:55:48|
|130.255.202.138| 6.066112278135983|2023-07-07 22:26:15|
| 198.33.206.140|1.9147905257021836|2023-03-01 04:44:14|
| 84.183.253.20| 7.707176860385722|2021-08-26 23:24:31|
|218.163.165.232| 9.458673015973213|2021-02-22 12:13:15|
| 62.57.20.153|1.5764916247359229|2021-11-06 12:41:59|
| 98.171.202.249| 3.546118349483626|2022-07-05 10:55:26|
|180.140.248.193|0.9512956363005021|2021-06-27 18:16:58|
| 13.236.170.177| 2.41361938344535|2021-08-11 02:19:06|
| 26.140.88.127| 7.51335778127692|2023-06-02 14:13:30|
| 7.118.207.252| 6.450499049816286|2022-12-11 06:36:20|
| 11.8.10.136| 8.750329246667354|2023-02-03 05:33:16|
| 232.140.56.86| 4.289740988237201|2023-02-22 20:10:09|
| 68.117.9.255| 5.384340363304169|2022-12-03 09:55:26|
+---------------+------------------+-------------------+
20
16
fre. 15. sep. 2023 kl. 20:14 skrev ashok34668@yahoo.com.INVALID
<as...@yahoo.com.invalid>:
> Hi team,
>
> I am using PySpark 3.4
>
> I have a table of million rows that has few columns. among them incoming
> ips and what is known as gbps (Gigabytes per second) and date and time
> of incoming ip.
>
> I want to filter out 20% of low active ips and work on the remainder of
> data. How can I do thiis in PySpark?
>
> Thanks
>
>
>
>
>
--
Bjørn Jørgensen
Vestre Aspehaug 4, 6010 Ålesund
Norge
+47 480 94 297