DEV Community: Raju Nekadi

How to write memory efficient machine learning model prediction data pipelines in Python,With an Example

Raju Nekadi — Fri, 10 May 2024 17:57:09 +0000

Introduction

If you are machine learning engineer and started to productionalize your model prediction pipeline you may came across below question.

how to write memory efficient prediction data pipelines in python.

Working with a dataset that is too large to fit into memory

Then this post is for you. I will go over how to write memory efficient data pipelines using generators and batching.

Let's setup dataset for our example, here we will be NYC Parking Violations Issued - Fiscal Year 2021,which contains violations issued during the 2014 fiscal year. This dataset contains several data quality problems and is a great example for demonstrating data profiling and data cleaning steps.

The dataset consists of over 15 million rows and the compressed data file is about 500 MB in size (~2.7 GB uncompressed). The dataset is available for download via the Socrata Open Data API (SODA). To download the dataset, openclean includes a SODA wrapper that provides access to all datasets available via the API using their unique identifier.

# List all datasets from NYC Open Data that contain 'parking'
# and ' violation'  in their name.

from openclean.data.source.socrata import Socrata

for dataset in Socrata().catalog(domain='data.cityofnewyork.us'):
    if 'parking' in dataset.name.lower() and 'violation' in dataset.name.lower():
        print(f'{dataset.identifier}\t{dataset.domain}\t{dataset.name}')

#Output

nc67-uf89   data.cityofnewyork.us   Open Parking and Camera Violations
pvqr-7yc4   data.cityofnewyork.us   Parking Violations Issued - Fiscal Year 2022
ncbg-6agr   data.cityofnewyork.us   DOF Parking Violation Codes
jt7v-77mi   data.cityofnewyork.us   Parking Violations Issued - Fiscal Year 2014
kiv2-tbus   data.cityofnewyork.us   Parking Violations Issued - Fiscal Year 2016
2bnn-yakx   data.cityofnewyork.us   Parking Violations Issued - Fiscal Year 2017
c284-tqph   data.cityofnewyork.us   Parking Violations Issued - Fiscal Year 2015
a5td-mswe   data.cityofnewyork.us   Parking Violations Issued - Fiscal Year 2018
faiq-9dfq   data.cityofnewyork.us   Parking Violations Issued - Fiscal Year 2019
p7t3-5i9s   data.cityofnewyork.us   Parking Violations Issued - Fiscal Year 2020
kvfd-bves   data.cityofnewyork.us   Parking Violations Issued - Fiscal Year 2021

Dataset

The parking violations dataset has the unique identifier kvfd-bves. The identifier is part of the dataste Url https://data.cityofnewyork.us/City-Government/Parking-Violations-Issued-Fiscal-Year-2021/kvfd-bves. The following code downloads the dataset in tab-delimited CSV format and stores it in a local file called kvfd-bves.tsv.gz.

What is Generators and How you you can use it?

In Python, a generator is a function that returns an iterator that produces a sequence of values when iterated over.

Generators are useful when we want to produce a large sequence of values, but we don't want to store all of them in memory at once.

Creating a generator is very similar to creating a function. You can use the generator expression () or define a function that uses yield instead of return.

Generators are usually used in a loop. For each iteration of the calling loop:

Control is passed back to the generator function from the calling loop.
The generator yields the next value to the loop and the control is passed back to the loop.
Steps 1 and 2 are repeated until all the generated values are exhausted.

You can also get data directly from a generator using next(generator_obj).

Let’s say our goal is to process the parking violation 2021 dataset with the following steps:

Keep only the violations issued by police (denoted by P in the data), to vehicles with the make TOYOTA in NJ.
Replace P with police.
Concat house number, street name, and registration state fields into a single address field.
Write the result’s in to a csv file with the header vehicle_make,issuing_agency,address.

Using Generator Expression ()

We will create generators with the () comprehension format. We also chain multiple generators to form a data pipeline, this is called chaining.

import csv


#define your input and output datasets
input_file_name = "./parking-violations-year-2021.csv"
output_file_name = "./nj_toyota_police_2021.csv"


# 1. stream data from input file
read_file_csv = open(input_file_name, "r")
extractor = csv.reader(read_file_csv)  # csv reader produces a generator

# 2. keep only required fields
# field index => field;
# 2 => registration state, 7 => vehicle make, 8 => issuing agency, 23 => house number, 24 => street name
col_filtered_df = ([row[2], row[7], row[8], row[23], row[24]] for row in extractor)

# 3. keep only violations issued by police, to vehicles with the make FORD in NJ
value_filtered_df = filter(
    lambda x: all([x[0] == "NJ", x[1] == "FORD", x[2] == "P"]), col_filtered_df
)

# 4. replace P with police
transformed_df = (
    [op[0], op[1], "Police", op[3], op[4]]
    for op in value_filtered_df
)

# 5. concat house number, street name, registration state into a single address field
final_df = (
    [op[1], op[2], ", ".join([op[3], op[4], op[1]])]
    for op in transformed_df
)

final_df  # this is a generator object and has not yet started generating data

# 6. write a header row for output data
write_file_object = open(output_file_name, "w")
loader = csv.writer(
    write_file_object, delimiter=",", quotechar='"', quoting=csv.QUOTE_MINIMAL
)
header = ["vehicle_make", "issuing_agency", "address"]
loader.writerow(header)

# 7. stream data into an output file
loader.writerows(final_df)  # loader asks for data from final_stream

Using generator yield

You can use yield statement to create your own generator inside function as shown below:


file_path = "./parking-violations-year-2021.csv"
def gen_file_reader(file_path):
    for row in open(file_path, "r"):
         yield row

def gen_print_row_count():
    count = 0
    for row in gen_file_reader("large_file"):
        count += 1
    print(f"Total count is {count}")

gen_print_row_count()

This is a simple example, but generator function is more memory efficient.

Batching

Consider an example where your data pipeline has to call an external paid API service. Let’s assume that this service charges per call. It does not matter if you call it with 1 row vs 10,000 rows, It charges the same price.

In this scenario you can accumulate the rows in memory. When it hits a specified threshold, then make the API call.

import time
import csv

input_file_name = "./parking-violations-year-2021.csv"
output_file_name = "./api_call_enriched.csv"


def api_call(rows):
    time.sleep(10)  # simulating an api call
    return rows


def batched_api_transforms(rows, batch_size=10000):
    batch = []
    for row in rows:
        batch.append(row)
        if len(batch) >= batch_size:
            yield from api_call(batch)
            batch = []
    yield from api_call(batch)


# 1. stream data from input file
read_file_object = open(input_file_name, "r")
extractor = csv.reader(read_file_object)

# 2. make batched calls to the external service
final_df = batched_api_transforms(extractor)
next(final_df)  # you will notice a 15s wait, simulating the external service call
[
    next(final_df) for i in range(9999)
]  # you will notice almost no wait, since this data is held in process memory

# we have iterated through the first batch of 10,000
# the next call will invoke the api_call function, thus sleeping for 15s
next(final_df)  # you will notice a 10s wait

Conclusion

In this post we saw how we can use generators and batching to process large
dataset, so next time you have to build a data pipeline to process a larger than memory data set, try using generators and batching.

Some Final Words
If this blog was helpful and you wish to show a little support, you could:

👍 300 times for this story
Follow me on LinkedIn: https://www.linkedin.com/in/raju-n-203b2115/

These actions really really really help me out, and are much appreciated!

Few new things in Python which I learned last week.

Raju Nekadi — Mon, 06 May 2024 22:03:56 +0000

Chunkify Huge List into Smaller N equal size lists.

In order to backfill data for one of our machine learning pipeline I have to divide the date list into small n list of equal length and distribute them at n GPU cluster.

from datetime import timedelta,date,datetime
start_dt = date(2023,1,1)
end_dt = date(2023,12,31)
cdays = []

while start_dt < end_dt:
    cdays.append(start_dt)
    start_dt += timedelta(days=3)

#print(cdays)

def split(a, n):
    k, m = divmod(len(a), n)
    return (a[i*k+min(i, m):(i+1)*k+min(i+1, m)] for i in range(n))

split_list = list(split(cdays, 15))
print(split_list[0])

#Output
[datetime.date(2023, 1, 1), datetime.date(2023, 1, 4), datetime.date(2023, 1, 7), datetime.date(2023, 1, 10), datetime.date(2023, 1, 13), datetime.date(2023, 1, 16), datetime.date(2023, 1, 19), datetime.date(2023, 1, 22), datetime.date(2023, 1, 25)]

__all__ in Python

The __all__ represents name of variable ,function or module that you want to expose to wildcards import. This method really comes handy when large number of function in your base module and you just want to export only few of them. Example let's say you have.


#foo.py

waz = 3
bar = 9

def baz():
    return 'baz'

__all__ = ['bar','baz']

and now bar.py imports from foo.py

#bar.py

from foo import *

print(bar)

print(baz)

print(waz) # This will trigger the exception , as waz was is not exported by the module.

Conclusion

Hope this was easy and clear to understand.

Some Final Words
If this blog was helpful and you wish to show a little support, you could:

👍 300 times for this story
Follow me on LinkedIn: https://www.linkedin.com/in/raju-n-203b2115/

These actions really really really help me out, and are much appreciated!

Scheduling a BigQuery SQL script, using Apache Airflow, with an example

Raju Nekadi — Fri, 03 May 2024 18:34:32 +0000

In my day to day work one of the most common use cases for Apache Airflow is to run hundreds of scheduled BigQuery SQL scripts. Developers who start with Airflow often ask the following questions

How to use airflow to orchestrate sql?

This post aims to cover the above questions. This post assumes you have a basic understanding of Apache Airflow , Python and BigQuery SQL.

Let’s assume we want to run a BigQuery SQL script every day at midnight.

DAG: Directed Acyclic Graph, In Airflow this is used to denote a data pipeline which runs on a scheduled interval. A DAG can be made up of one or more individual tasks.

from airflow import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
 
default_arg = {'owner': 'raju', 'start_date': '2024-04-29'}
 
dag = DAG('simple-bigquery-dag',
          default_args=default_arg,
          schedule_interval='0 0 * * *')
 
bqsql_task = BigQueryInsertJobOperato(dag=dag,
                           gcp_conn_id ='bqsql_default', 
                           task_id='bqsql_task'
                           configuration={"query": {
                               "query": "{% include 
                           'sql/bqsql_sample_query.sql' %}",
                            "useLegacySql": False,}},
                           params={'project_id': MYPROJECT_ID,
                           'dataset_name': MYDATASET_NAME,
        }

In the above script

0 0 * * * is a cron schedule format, denoting that the DAG should be run everyday at midnight, which is denoted by the 0th hour of every day. (note that Airflow by default runs on UTC time) gcp_conn_id is the connection id for your BigQuery SQL database, you can set this in admin -> connections from airflow UI. There you will set the username and password that Airflow uses to access your database. ‍

The SQL script to perform this operation is stored in a separate file bqsql_sample_query.sql. This file is read in when the DAG is being run.

CREATE TABLE IF NOT EXISTS {{params.project_id}}.{{params.dataset_name}}.Inventory
INSERT {{params.project_id}}.{{params.dataset_name}}.Inventory (product, quantity)
VALUES('top load washer', 10),
      ('front load washer', 20),
      ('dryer', 30),
      ('refrigerator', 10),
      ('microwave', 20),
      ('dishwasher', 30),
      ('oven', 5);

There are 2 key concepts in the templated SQL script shown above

Airflow macros: They provide access to the metadata that is available for each DAG run. We use the execution date as it provides the previous date over which we want to aggregate the data. ref: https://airflow.apache.org/docs/stable/macros.html

Templated parameters: If we want our SQL script to have some parameters that can be filled at run time from the DAG, we can pass them as parameters to the task. In our example we passed the project_id (MYDATASET_NAME).

Conclusion

In this post we saw

How to schedule BigQuery SQL scripts using Apache Airflow
How Airflow connects to the database using the connection id
How to pass in parameters at run-time using input parameters and macros. Hope this gives you an understanding of how to schedule BigQuery SQL scripts in Airflow and how to use templating.

Some Final Words
If this blog was helpful and you wish to show a little support, you could:

👍 300 times for this story
Follow me on LinkedIn: https://www.linkedin.com/in/raju-n-203b2115/

These actions really really really help me out, and are much appreciated!

How “If name == ‘main' ’” Works?

Raju Nekadi — Mon, 29 Apr 2024 18:48:20 +0000

Solving the mystery of If __name__ == '__main__'.

Let's assume you have a python script first.py

# first.py

print('Hola from first.py')

if __name__ == '__main__':
    print('Hola first module __name__:' __name__')

# output
# Hola from first.py
# Hola first module __name__: __main__

If we run first.py using the command Python
python first.py, the __name__ variable for first.py will automatically become __main__.

# second.py

print('Hola from second.py')

if __name__ == '__main__':
    print('Hola second module __name__:' __name__')

# output
# Hola from second.py
# Hola second module __name__: __main__

Similarly, if we run second.py using the command python second.py, the __name__ variable for second.py will automatically become __main__

What if we have 2.py files

#second.py

def hola_second():
    print('hola_second__name__:',__name__)

# first.py

from second import hola_second
hola_second()

def hola_first():
    print('hola_first__name__:',__name__)

# python first.py
# hola_second__name__: second
# hola_first__name__: __main__

And in first.py, we import a function from second.py. And in this function in second.py, we print __name__. And we run first.py directly using python first.py (note that second.py is not run directly)
Since we run first.py directly, only __name__ in first.py will be '__main__'. The __name__ imported from second.py will be 'second' instead of '__main__'.

Now what does adding if __name__ == ‘__main__’ do

# second.py

def hola_second():
  print('from second.py:', __name__)

if __name__ == '__main__':
  print('I am second')


# first.py

from second import hola_second
hola_second()

Notice that I am second is no longer printed.
And this is because of the if __name__ == '__main__' statement we’ve put in second.py. Whatever is inside this if block will only run if we run the Python script directly.

In this case, print('I am second') will only run if we run second.py directly, meaning that we use the command python second.py.

Conclusion
if __name__ == ‘__main__’ is used as conditional statement to allow only run code that we want to run.

Some Final Words
If this blog was helpful and you wish to show a little support, you could:

1. 👍 300 times for this story
2. Follow me on LinkedIn: https://www.linkedin.com/in/raju-n-203b2115/

These actions really really really help me out, and are much appreciated!

DEV Community: Raju Nekadi

How to write memory efficient machine learning model prediction data pipelines in Python,With an Example

Few new things in Python which I learned last week.

Scheduling a BigQuery SQL script, using Apache Airflow, with an example

How “If __name__ == ‘__main__' ’” Works?

How “If name == ‘main' ’” Works?