How to parallelize for loop with multiple functions inside - Python - AWS Glue

user_19846134 0
Viewed 75

Here is my question.

I have created several functions, and those functions will be run per each UserID (that is the reason of the for loop below running the functions). It will be run in AWS Glue.

I need to do scalable this code in Python / AWS Glue, working with millions of UsersID. Since each UserID has 3000 records, then 1 million Users will be 3000000000 records in total.

I thought about parallelize the for loop below, and I read about Dask, Pandarell, and some others. (Even increasing resources in AWS Glue, like worker types, but not working)

The problem is, I do not know how to implement those libraries when I have to run a for loop with several functions inside, and when the output of every functions is the input of the following function.

Does anyone have a clue of how can I parallelize that loop? (Could be using Python, Spark, PySpark, etc)

Thanks in advance

raw_data = pd.read_csv('C:{path}/tracking.csv') 
raw_data["Time"]=pd.to_datetime(raw_data.Time)
raw_data=raw_data.sort_values(['UserId', 'Time'], ascending=[True, True])

listIdUser = raw_data['UserId'].unique().tolist() 
poi = pd.DataFrame(columns=['PointsId', 'TimeInitial', 'TimeEnding', 'Lat', 'Lon', 'TotalTime', 'UserId'])

for i in listIdUser:
    source_data = raw_data[raw_data["UserId"] == i]
    source_data=source_data.reset_index().drop(["index"], axis=1)
    clean_data = single_outlier_detection(source_data)
    pairwise_distance = pairwise_distance_calculation(clean_data)
    data = consecutive_points(clean_data)
    dbscan_data=coordinates_data_preparation(data)
    distances = distances_estimation(dbscan_data)
    threshold=dbscan_epsilon_estimation(distances,pairwise_distance)
    cluster_df = dbscan_model_run(threshold, dbscan_data)
    points_stay = places_of_interest(cluster_df)
    poi=pd.concat([poi, points_stay])
aws-glue pyspark amazon-web-services python
1 Answers

Since your logic is already defined for one listIdUser, all you have to do is wrap this logic with Fugue. We can separate the partitioning and execution. I don't have data to test on but it will look like this.

raw_data = pd.read_csv('C:{path}/tracking.csv') 
raw_data["Time"]=pd.to_datetime(raw_data.Time)

def logic_one_id(df: pd.DataFrame) -> pd.DataFrame:
    # notice I remove the step to subset raw data
    source_data=df.reset_index().drop(["index"], axis=1)
    clean_data = single_outlier_detection(source_data)
    pairwise_distance = pairwise_distance_calculation(clean_data)
    data = consecutive_points(clean_data)
    dbscan_data=coordinates_data_preparation(data)
    distances = distances_estimation(dbscan_data)
    threshold=dbscan_epsilon_estimation(distances,pairwise_distance)
    cluster_df = dbscan_model_run(threshold, dbscan_data)
    points_stay = places_of_interest(cluster_df)
    # notice I remove the concat. Fugue will handle
    return points_stay

and then you can do:

output_schema="PointsId:int,TimeInitial:datetime,TimeEnding:datetime,Lat:float,Lon:float,TotalTime:float,UserId:int"

from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()

from fugue import transform
transform(df, logic_one_id, schema=output_schema, partition={"by":"UserId", "presort":"Time"}, engine=spark)

A couple of things to notice:

  1. Schema is a requirement for Spark so I guessed your output schema
  2. The transform() function will handle the partition by UserID so you don't need to do it yourself
  3. We presorted the data by time also during partitioning
  4. We passed in spark as the engine. If engine=None, it will run on Pandas so you can test locally pretty easily.

You can test on Pandas like this:

test_df = raw_data[raw_data["UserId"] == one_id_here]
transform(test_df, logic_one_id, schema=output_schema, partition={"by":"UserId", "presort":"Time"})

and if it works, run all the data on the spark engine

user_111632140
Related